Assembly.js

import Emitter from 'component-emitter'
import has from '@uppy/utils/lib/hasProperty'
import NetworkError from '@uppy/utils/lib/NetworkError'
import fetchWithNetworkError from '@uppy/utils/lib/fetchWithNetworkError'
import parseUrl from './parseUrl.js'

// We used to lazy load socket.io to avoid a console error
// in IE 10 when the Transloadit plugin is not used.
// (The console.error call comes from `buffer`. I
// think we actually don't use that part of socket.io
// at all…)
// TODO: remove this hack in the next release.
let socketIo
function requireSocketIo () {
  // eslint-disable-next-line no-return-assign, no-restricted-globals, global-require, no-undef
  return socketIo ??= require('socket.io-client')
}

const ASSEMBLY_UPLOADING = 'ASSEMBLY_UPLOADING'
const ASSEMBLY_EXECUTING = 'ASSEMBLY_EXECUTING'
const ASSEMBLY_COMPLETED = 'ASSEMBLY_COMPLETED'

const statusOrder = [
  ASSEMBLY_UPLOADING,
  ASSEMBLY_EXECUTING,
  ASSEMBLY_COMPLETED,
]

/**
 * Check that an assembly status is equal to or larger than some desired status.
 * It checks for things that are larger so that a comparison like this works,
 * when the old assembly status is UPLOADING but the new is FINISHED:
 *
 * !isStatus(oldStatus, ASSEMBLY_EXECUTING) && isStatus(newState, ASSEMBLY_EXECUTING)
 *
 * …so that we can emit the 'executing' event even if the execution step was so
 * fast that we missed it.
 */
function isStatus (status, test) {
  return statusOrder.indexOf(status) >= statusOrder.indexOf(test)
}

class TransloaditAssembly extends Emitter {
  #rateLimitedQueue

  #fetchWithNetworkError

  #previousFetchStatusStillPending = false

  constructor (assembly, rateLimitedQueue) {
    super()

    // The current assembly status.
    this.status = assembly
    // The socket.io connection.
    this.socket = null
    // The interval timer for full status updates.
    this.pollInterval = null
    // Whether this assembly has been closed (finished or errored)
    this.closed = false

    this.#rateLimitedQueue = rateLimitedQueue
    this.#fetchWithNetworkError = rateLimitedQueue.wrapPromiseFunction(fetchWithNetworkError)
  }

  connect () {
    this.#connectSocket()
    this.#beginPolling()
  }

  #onFinished () {
    this.emit('finished')
    this.close()
  }

  #connectSocket () {
    const parsed = parseUrl(this.status.websocket_url)
    const socket = requireSocketIo().connect(parsed.origin, {
      transports: ['websocket'],
      path: parsed.pathname,
    })

    socket.on('connect', () => {
      socket.emit('assembly_connect', {
        id: this.status.assembly_id,
      })

      this.emit('connect')
    })

    socket.on('connect_error', () => {
      socket.disconnect()
      this.socket = null
    })

    socket.on('assembly_finished', () => {
      this.#onFinished()
    })

    socket.on('assembly_upload_finished', (file) => {
      this.emit('upload', file)
      this.status.uploads.push(file)
    })

    socket.on('assembly_uploading_finished', () => {
      this.emit('executing')
    })

    socket.on('assembly_upload_meta_data_extracted', () => {
      this.emit('metadata')
      this.#fetchStatus({ diff: false })
    })

    socket.on('assembly_result_finished', (stepName, result) => {
      this.emit('result', stepName, result)
      if (!this.status.results[stepName]) {
        this.status.results[stepName] = []
      }
      this.status.results[stepName].push(result)
    })

    socket.on('assembly_error', (err) => {
      this.#onError(err)
      // Refetch for updated status code
      this.#fetchStatus({ diff: false })
    })

    this.socket = socket
  }

  #onError (err) {
    this.emit('error', Object.assign(new Error(err.message), err))
    this.close()
  }

  /**
   * Begin polling for assembly status changes. This sends a request to the
   * assembly status endpoint every so often, if the socket is not connected.
   * If the socket connection fails or takes a long time, we won't miss any
   * events.
   */
  #beginPolling () {
    this.pollInterval = setInterval(() => {
      if (!this.socket || !this.socket.connected) {
        this.#fetchStatus()
      }
    }, 2000)
  }

  /**
   * Reload assembly status. Useful if the socket doesn't work.
   *
   * Pass `diff: false` to avoid emitting diff events, instead only emitting
   * 'status'.
   */
  async #fetchStatus ({ diff = true } = {}) {
    if (this.closed || this.#rateLimitedQueue.isPaused || this.#previousFetchStatusStillPending) return

    try {
      this.#previousFetchStatusStillPending = true
      const response = await this.#fetchWithNetworkError(this.status.assembly_ssl_url)
      this.#previousFetchStatusStillPending = false

      if (this.closed) return

      if (response.status === 429) {
        this.#rateLimitedQueue.rateLimit(2_000)
        return
      }

      if (!response.ok) {
        this.#onError(new NetworkError(response.statusText))
        return
      }

      const status = await response.json()
      // Avoid updating if we closed during this request's lifetime.
      if (this.closed) return
      this.emit('status', status)

      if (diff) {
        this.updateStatus(status)
      } else {
        this.status = status
      }
    } catch (err) {
      this.#onError(err)
    }
  }

  update () {
    return this.#fetchStatus({ diff: true })
  }

  /**
   * Update this assembly's status with a full new object. Events will be
   * emitted for status changes, new files, and new results.
   *
   * @param {object} next The new assembly status object.
   */
  updateStatus (next) {
    this.#diffStatus(this.status, next)
    this.status = next
  }

  /**
   * Diff two assembly statuses, and emit the events necessary to go from `prev`
   * to `next`.
   *
   * @param {object} prev The previous assembly status.
   * @param {object} next The new assembly status.
   */
  #diffStatus (prev, next) {
    const prevStatus = prev.ok
    const nextStatus = next.ok

    if (next.error && !prev.error) {
      return this.#onError(next)
    }

    // Desired emit order:
    //  - executing
    //  - (n × upload)
    //  - metadata
    //  - (m × result)
    //  - finished
    // The below checks run in this order, that way even if we jump from
    // UPLOADING straight to FINISHED all the events are emitted as expected.

    const nowExecuting = isStatus(nextStatus, ASSEMBLY_EXECUTING)
      && !isStatus(prevStatus, ASSEMBLY_EXECUTING)
    if (nowExecuting) {
      // Without WebSockets, this is our only way to tell if uploading finished.
      // Hence, we emit this just before the 'upload's and before the 'metadata'
      // event for the most intuitive ordering, corresponding to the _usual_
      // ordering (if not guaranteed) that you'd get on the WebSocket.
      this.emit('executing')
    }

    // Find new uploaded files.
    Object.keys(next.uploads)
      .filter((upload) => !has(prev.uploads, upload))
      .forEach((upload) => {
        this.emit('upload', next.uploads[upload])
      })

    if (nowExecuting) {
      this.emit('metadata')
    }

    // Find new results.
    Object.keys(next.results).forEach((stepName) => {
      const nextResults = next.results[stepName]
      const prevResults = prev.results[stepName]

      nextResults
        .filter((n) => !prevResults || !prevResults.some((p) => p.id === n.id))
        .forEach((result) => {
          this.emit('result', stepName, result)
        })
    })

    if (isStatus(nextStatus, ASSEMBLY_COMPLETED)
        && !isStatus(prevStatus, ASSEMBLY_COMPLETED)) {
      this.emit('finished')
    }

    return undefined
  }

  /**
   * Stop updating this assembly.
   */
  close () {
    this.closed = true
    if (this.socket) {
      this.socket.disconnect()
      this.socket = null
    }
    clearInterval(this.pollInterval)
    this.pollInterval = null
  }
}

export default TransloaditAssembly