ci-visibility-intake.js

'use strict'

const http = require('http')
const zlib = require('zlib')
const express = require('express')
const bodyParser = require('body-parser')
const msgpack = require('@msgpack/msgpack')
const multer = require('multer')
const upload = multer()

const FakeAgent = require('./helpers/fake-agent')

const DEFAULT_SETTINGS = {
  code_coverage: true,
  tests_skipping: true,
  itr_enabled: true,
  require_git: false,
  early_flake_detection: {
    enabled: false,
    slow_test_retries: {
      '5s': 3,
    },
  },
  flaky_test_retries_enabled: false,
  di_enabled: false,
  known_tests_enabled: false,
  test_management: {
    enabled: false,
  },
  impacted_tests_enabled: false,
  coverage_report_upload_enabled: false,
}

const DEFAULT_SUITES_TO_SKIP = []
const DEFAULT_SKIPPABLE_COVERAGE = {}
const DEFAULT_GIT_UPLOAD_STATUS = 200
const DEFAULT_KNOWN_TESTS_RESPONSE_STATUS = 200
const DEFAULT_INFO_RESPONSE = {
  endpoints: ['/evp_proxy/v2', '/debugger/v1/input'],
}
const DEFAULT_CORRELATION_ID = '1234'
const DEFAULT_KNOWN_TESTS = ['test-suite1.js.test-name1', 'test-suite2.js.test-name2']

const DEFAULT_TEST_MANAGEMENT_TESTS = {}
const DEFAULT_TEST_MANAGEMENT_TESTS_RESPONSE_STATUS = 200

class FakeCiVisIntake extends FakeAgent {
  #settings = DEFAULT_SETTINGS
  #settingsResponses = []
  #settingsResponseDelayMs = 0
  #settingsResponseStatusCode = 200
  #settingsResponseStatusCodes = []
  #mediaResponseDelayMs = 0
  #mediaResponsesPending = false
  #mediaResponseStatusCode = 201
  #suitesToSkip = DEFAULT_SUITES_TO_SKIP
  #skippableCoverage = DEFAULT_SKIPPABLE_COVERAGE
  #gitUploadStatus = DEFAULT_GIT_UPLOAD_STATUS
  #infoResponse = DEFAULT_INFO_RESPONSE
  #correlationId = DEFAULT_CORRELATION_ID
  #knownTests = DEFAULT_KNOWN_TESTS
  #knownTestsStatusCode = DEFAULT_KNOWN_TESTS_RESPONSE_STATUS
  #waitingTime = 0
  #knownTestsPageIndex = 0
  #testManagementResponse = DEFAULT_TEST_MANAGEMENT_TESTS
  #testManagementResponses = []
  #testManagementResponseStatusCode = DEFAULT_TEST_MANAGEMENT_TESTS_RESPONSE_STATUS
  #skippableSuitesResponseStatusCode = 200

  #getSkippableResponse () {
    const meta = { correlation_id: this.#correlationId }
    if (Object.keys(this.#skippableCoverage).length) {
      meta.coverage = this.#skippableCoverage
    }
    return { data: this.#suitesToSkip, meta }
  }

  setKnownTestsResponseCode (statusCode) {
    this.#knownTestsStatusCode = statusCode
  }

  setKnownTests (newKnownTestsResponse) {
    this.#knownTests = newKnownTestsResponse
  }

  setInfoResponse (newInfoResponse) {
    this.#infoResponse = newInfoResponse
  }

  setGitUploadStatus (newStatus) {
    this.#gitUploadStatus = newStatus
  }

  setSuitesToSkip (newSuitesToSkip) {
    this.#suitesToSkip = newSuitesToSkip
  }

  setSkippableCoverage (newSkippableCoverage) {
    this.#skippableCoverage = newSkippableCoverage
  }

  setItrCorrelationId (newCorrelationId) {
    this.#correlationId = newCorrelationId
  }

  setSettings (newSettings) {
    this.#settings = newSettings
  }

  /**
   * Sets library configuration responses to return in order.
   *
   * @param {object[]} responses
   * @returns {void}
   */
  setSettingsResponses (responses) {
    this.#settingsResponses = responses.slice()
  }

  /**
   * Delays settings responses to exercise initialization ordering.
   *
   * @param {number} delayMs
   * @returns {void}
   */
  setSettingsResponseDelay (delayMs) {
    this.#settingsResponseDelayMs = delayMs
  }

  setSettingsResponseCode (statusCode) {
    this.#settingsResponseStatusCode = statusCode
  }

  /**
   * @param {number[]} statusCodes
   */
  setSettingsResponseStatusCodes (statusCodes) {
    this.#settingsResponseStatusCodes = statusCodes.slice()
  }

  // Lets a test simulate the media endpoint failing (e.g. 500) to verify the
  // cypress run still completes and reports normally when an upload fails.
  setMediaResponseStatusCode (statusCode) {
    this.#mediaResponseStatusCode = statusCode
  }

  /**
   * @param {number} delayMs - Delay before responding to screenshot uploads
   * @returns {void}
   */
  setMediaResponseDelay (delayMs) {
    this.#mediaResponseDelayMs = delayMs
  }

  /**
   * Leaves media requests open until the client cancels them.
   *
   * @returns {void}
   */
  setMediaResponsesPending () {
    this.#mediaResponsesPending = true
  }

  setWaitingTime (newWaitingTime) {
    this.#waitingTime = newWaitingTime
  }

  setTestManagementTests (newTestManagementTests) {
    this.#testManagementResponse = newTestManagementTests
  }

  /**
   * Sets Test Management responses to return in order.
   *
   * @param {object[]} responses
   * @returns {void}
   */
  setTestManagementTestResponses (responses) {
    this.#testManagementResponses = responses.slice()
  }

  setTestManagementTestsResponseCode (newStatusCode) {
    this.#testManagementResponseStatusCode = newStatusCode
  }

  setSkippableSuitesResponseCode (statusCode) {
    this.#skippableSuitesResponseStatusCode = statusCode
  }

  async start () {
    const app = express()
    app.use(bodyParser.raw({ limit: Infinity, type: 'application/msgpack' }))

    const handleV04Traces = (req, res) => {
      if (req.body.length === 0) return res.status(200).send()
      res.status(200).send({ rate_by_service: { 'service:,env:': 1 } })
      this.emit('message', {
        headers: req.headers,
        payload: msgpack.decode(req.body, { useBigInt64: true }),
        url: req.url,
      })
    }
    app.put('/v0.4/traces', handleV04Traces)
    app.post('/v0.4/traces', handleV04Traces)

    app.get('/info', (req, res) => {
      res.status(200).send(JSON.stringify(this.#infoResponse))
      this.emit('message', {
        headers: req.headers,
        url: req.url,
      })
    })

    // It can be slowed down with setWaitingTime
    app.post(['/api/v2/citestcycle', '/evp_proxy/:version/api/v2/citestcycle'], (req, res) => {
      this.waitingTimeoutId = setTimeout(() => {
        res.status(200).send('OK')
        this.emit('message', {
          headers: req.headers,
          payload: msgpack.decode(req.body, { useBigInt64: true }),
          url: req.url,
        })
      }, this.#waitingTime || 0)
    })

    app.post([
      '/api/v2/git/repository/search_commits',
      '/evp_proxy/:version/api/v2/git/repository/search_commits',
    ], (req, res) => {
      res.status(this.#gitUploadStatus).send(JSON.stringify({ data: [] }))
      this.emit('message', {
        headers: req.headers,
        payload: req.body,
        url: req.url,
      })
    })

    app.post([
      '/api/v2/git/repository/packfile',
      '/evp_proxy/:version/api/v2/git/repository/packfile',
    ], (req, res) => {
      res.status(202).send('')
      this.emit('message', {
        headers: req.headers,
        url: req.url,
      })
    })

    app.post([
      '/api/v2/citestcov',
      '/evp_proxy/:version/api/v2/citestcov',
    ], upload.any(), (req, res) => {
      res.status(200).send('OK')

      const coveragePayloads = req.files
        .filter((file) => file.fieldname !== 'event')
        .map((file) => {
          return {
            name: file.fieldname,
            type: file.mimetype,
            filename: file.originalname,
            content: msgpack.decode(file.buffer),
          }
        })

      this.emit('message', {
        headers: req.headers,
        payload: coveragePayloads,
        url: req.url,
      })
    })

    // Coverage report upload endpoint - one file per request
    app.post([
      '/api/v2/cicovreprt',
      '/evp_proxy/:version/api/v2/cicovreprt',
    ], upload.any(), (req, res) => {
      res.status(200).send('OK')

      const coverageFile = req.files.find(f => f.fieldname === 'coverage')
      const eventFile = req.files.find(f => f.fieldname === 'event')

      this.emit('message', {
        headers: req.headers,
        coverageFile: coverageFile && {
          name: coverageFile.fieldname,
          content: zlib.gunzipSync(coverageFile.buffer).toString('utf8'),
        },
        eventFile: eventFile && {
          name: eventFile.fieldname,
          content: JSON.parse(eventFile.buffer.toString('utf8')),
        },
        url: req.url,
      })
    })

    app.post('/api/v2/ci/test-runs/:traceId/media', express.raw({ limit: Infinity, type: '*/*' }), (req, res) => {
      const receivedAtMs = Date.now()
      const respond = () => {
        res.status(this.#mediaResponseStatusCode).send()
        this.emit('message', {
          headers: req.headers,
          media: {
            traceId: req.params.traceId,
            contentType: req.headers['content-type'],
            // Metadata is carried as query params (not X-Dd-* headers) so it survives the Agent's
            // evp_proxy, which forwards only an allow-listed header set.
            idempotencyKey: req.query.idempotency_key,
            capturedAt: req.query.captured_at_ms,
            content: req.body,
            receivedAtMs,
          },
          url: req.url,
        })
      }

      if (this.#mediaResponsesPending) {
        return
      }
      if (this.#mediaResponseDelayMs > 0) {
        setTimeout(respond, this.#mediaResponseDelayMs)
      } else {
        respond()
      }
    })

    app.post([
      '/api/v2/libraries/tests/services/setting',
      '/evp_proxy/:version/api/v2/libraries/tests/services/setting',
    ], (req, res) => {
      const respond = () => {
        const settingsResponseStatusCode = this.#settingsResponseStatusCodes.shift() ??
          this.#settingsResponseStatusCode
        const settings = this.#settingsResponses.length
          ? this.#settingsResponses.shift()
          : this.#settings
        res.status(settingsResponseStatusCode)
        if (settingsResponseStatusCode >= 200 && settingsResponseStatusCode < 300) {
          res.send(JSON.stringify({
            data: {
              attributes: settings,
            },
          }))
        } else {
          res.send(JSON.stringify({ errors: ['error'] }))
        }
        this.emit('message', {
          headers: req.headers,
          url: req.url,
        })
      }

      if (this.#settingsResponseDelayMs > 0) {
        setTimeout(respond, this.#settingsResponseDelayMs)
      } else {
        respond()
      }
    })

    app.post([
      '/api/v2/ci/tests/skippable',
      '/evp_proxy/:version/api/v2/ci/tests/skippable',
    ], express.json(), (req, res) => {
      if (this.#skippableSuitesResponseStatusCode < 200 || this.#skippableSuitesResponseStatusCode >= 300) {
        res.status(this.#skippableSuitesResponseStatusCode).send(JSON.stringify({ errors: ['error'] }))
        return
      }
      res.status(this.#skippableSuitesResponseStatusCode).send(JSON.stringify(this.#getSkippableResponse()))
      this.emit('message', {
        headers: req.headers,
        payload: req.body,
        url: req.url,
      })
    })

    app.post([
      '/api/v2/ci/libraries/tests',
      '/evp_proxy/:version/api/v2/ci/libraries/tests',
    ], (req, res) => {
      // The endpoint returns compressed data if 'accept-encoding' is set to 'gzip'
      const isGzip = req.headers['accept-encoding'] === 'gzip'

      let responseData
      if (Array.isArray(this.#knownTests)) {
        // Paginated mode: knownTests is an array of page responses
        const page = this.#knownTestsPageIndex < this.#knownTests.length
          ? this.#knownTests[this.#knownTestsPageIndex]
          : null
        this.#knownTestsPageIndex++
        if (page) {
          responseData = JSON.stringify(page)
        } else {
          res.status(404).send('')
          return
        }
      } else {
        // Legacy single-response mode
        responseData = JSON.stringify({
          data: {
            attributes: {
              tests: this.#knownTests,
            },
          },
        })
      }

      res.setHeader('content-type', 'application/json')
      if (isGzip) {
        res.setHeader('content-encoding', 'gzip')
      }
      res.status(this.#knownTestsStatusCode).send(isGzip ? zlib.gzipSync(responseData) : responseData)
      this.emit('message', {
        headers: req.headers,
        url: req.url,
      })
    })

    app.post([
      '/api/v2/logs',
      '/debugger/v1/input',
    ], express.json(), (req, res) => {
      res.status(200).send('OK')
      this.emit('message', {
        headers: req.headers,
        url: req.url,
        logMessage: req.body,
      })
    })

    app.post([
      '/api/v2/test/libraries/test-management/tests',
      '/evp_proxy/:version/api/v2/test/libraries/test-management/tests',
    ], (req, res) => {
      res.setHeader('content-type', 'application/json')
      const testManagementResponse = this.#testManagementResponses.length
        ? this.#testManagementResponses.shift()
        : this.#testManagementResponse
      const data = JSON.stringify({
        data: {
          attributes: {
            modules: testManagementResponse,
          },
        },
      })
      res.status(this.#testManagementResponseStatusCode).send(data)
      this.emit('message', {
        headers: req.headers,
        url: req.url,
      })
    })

    app.post('/telemetry/proxy/api/v2/apmtelemetry', express.json(), (req, res) => {
      res.status(200).send()
      if (req.body?.payload?.namespace !== 'civisibility') return
      this.emit('message', {
        headers: req.headers,
        payload: req.body,
        url: req.url,
      })
    })

    return new Promise((resolve, reject) => {
      const timeoutObj = setTimeout(() => {
        reject(new Error('Intake timed out starting up'))
      }, 10000)
      this.server = http.createServer(app)
      this.server.on('error', reject)
      this.server.listen(this.port, () => {
        this.port = (/** @type {import('net').AddressInfo} */ (this.server.address())).port
        clearTimeout(timeoutObj)
        resolve(this)
      })
    })
  }

  resetKnownTestsPageIndex () {
    this.#knownTestsPageIndex = 0
  }

  stop () {
    this.#settings = DEFAULT_SETTINGS
    this.#settingsResponses = []
    this.#settingsResponseDelayMs = 0
    this.#settingsResponseStatusCode = 200
    this.#settingsResponseStatusCodes = []
    this.#suitesToSkip = DEFAULT_SUITES_TO_SKIP
    this.#skippableCoverage = DEFAULT_SKIPPABLE_COVERAGE
    this.#gitUploadStatus = DEFAULT_GIT_UPLOAD_STATUS
    this.#knownTestsStatusCode = DEFAULT_KNOWN_TESTS_RESPONSE_STATUS
    this.#knownTestsPageIndex = 0
    this.#infoResponse = DEFAULT_INFO_RESPONSE
    this.#mediaResponseDelayMs = 0
    this.#mediaResponsesPending = false
    this.#testManagementResponseStatusCode = DEFAULT_TEST_MANAGEMENT_TESTS_RESPONSE_STATUS
    this.#testManagementResponse = DEFAULT_TEST_MANAGEMENT_TESTS
    this.#testManagementResponses = []
    this.#skippableSuitesResponseStatusCode = 200
    this.removeAllListeners()
    if (this.waitingTimeoutId) {
      clearTimeout(this.waitingTimeoutId)
    }
    this.#waitingTime = 0
    return super.stop()
  }

  // Gather payloads while childProcess runs, then run onPayload once on the
  // accumulated buffer after the child emits `'exit'` plus `gracePeriod` ms of HTTP
  // drain. `hardTimeout` is a backstop for a genuinely hung child — bump it per-call
  // only when a workload's child runtime is provably above the default.
  /**
   * @param {import('child_process').ChildProcess | import('node:events').EventEmitter} childProcess
   *   Source of the `'exit'` event. `exitCode` / `signalCode` are read synchronously
   *   so a child that has already exited is handled correctly.
   * @param {(message: object) => boolean} [payloadMatch] Per-message filter; falsy
   *   accepts everything.
   * @param {(payloads: object[]) => void} onPayload Assertion callback, invoked once
   *   on the post-exit buffer. Callers can read `childProcess.exitCode` immediately
   *   after the returned promise resolves.
   * @param {{ gracePeriod?: number, hardTimeout?: number }} [options]
   */
  gatherPayloadsUntilChildExit (childProcess, payloadMatch, onPayload, options = {}) {
    const { gracePeriod = 1000, hardTimeout = 30_000 } = options
    const payloads = []

    return new Promise((resolve, reject) => {
      let settled = false
      let graceTimer = null

      const cleanup = () => {
        settled = true
        this.off('message', messageHandler)
        childProcess.off('exit', exitHandler)
        clearTimeout(hardTimer)
        clearTimeout(graceTimer)
      }

      const messageHandler = (message) => {
        if (settled) return
        if (!payloadMatch || payloadMatch(message)) {
          payloads.push(message)
        }
      }

      const exitHandler = () => {
        if (settled) return
        // Hung-child backstop is moot once the child has exited; only the
        // grace period applies from here. Without this, a child exiting
        // close to `hardTimeout` races the grace timer and rejects with a
        // wrong "child still running" message instead of running the
        // assertion.
        clearTimeout(hardTimer)
        graceTimer = setTimeout(() => {
          if (settled) return
          if (payloads.length === 0) {
            cleanup()
            reject(new Error(
              'gatherPayloadsUntilChildExit: child exited with no matching payloads ' +
              `(after ${gracePeriod}ms grace period)`
            ))
            return
          }
          try {
            onPayload(payloads)
            cleanup()
            resolve()
          } catch (error) {
            cleanup()
            reject(error)
          }
        }, gracePeriod)
      }

      const hardTimer = setTimeout(() => {
        if (settled) return
        cleanup()
        reject(new Error(
          `gatherPayloadsUntilChildExit: hard timeout of ${hardTimeout}ms expired (child still running)`
        ))
      }, hardTimeout)

      this.on('message', messageHandler)

      // Child may already have exited (very fast spawn-and-die).
      if (childProcess.exitCode !== null || childProcess.signalCode != null) {
        queueMicrotask(exitHandler)
      } else {
        childProcess.once('exit', exitHandler)
      }
    })
  }

  // Similar to gatherPayloads but resolves if enough payloads have been gathered
  // to make the assertions pass. It times out after maxGatheringTime so it should
  // always be faster or as fast as gatherPayloads
  gatherPayloadsMaxTimeout (payloadMatch, onPayload, maxGatheringTime = 15000) {
    const payloads = []
    return /** @type {Promise<void>} */ (new Promise((resolve, reject) => {
      const timeoutId = setTimeout(() => {
        try {
          onPayload(payloads)
          resolve()
        } catch (e) {
          reject(e)
        } finally {
          this.off('message', messageHandler)
        }
      }, maxGatheringTime)
      const messageHandler = (message) => {
        if (!payloadMatch || payloadMatch(message)) {
          payloads.push(message)
          try {
            onPayload(payloads)
            clearTimeout(timeoutId)
            this.off('message', messageHandler)
            resolve()
          } catch {
            // Assertion not yet satisfied — we'll try again when a new payload arrives.
            // The timeout handler will re-run onPayload and reject with the actual error.
          }
        }
      }
      this.on('message', messageHandler)
    }))
  }

  gatherPayloads (payloadMatch, gatheringTime = 15000) {
    const payloads = []
    return new Promise((resolve, reject) => {
      setTimeout(() => {
        this.off('message', messageHandler)
        if (payloads.length === 0) {
          reject(new Error('No payloads were received'))
        } else {
          resolve(payloads)
        }
      }, gatheringTime)
      const messageHandler = (message) => {
        if (!payloadMatch || payloadMatch(message)) {
          payloads.push(message)
        }
      }
      this.on('message', messageHandler)
    })
  }

  payloadReceived (payloadMatch, timeout) {
    return new Promise((resolve, reject) => {
      const timeoutId = setTimeout(() => {
        this.off('message', messageHandler)
        reject(new Error('Timeout'))
      }, timeout || 15000)
      const messageHandler = (message) => {
        if (!payloadMatch || payloadMatch(message)) {
          clearTimeout(timeoutId)
          resolve(message)
          this.off('message', messageHandler)
        }
      }
      this.on('message', messageHandler)
    })
  }

  assertPayloadReceived (fn, messageMatch, timeout) {
    return this.payloadReceived(messageMatch, timeout).then(fn)
  }
}

module.exports = { FakeCiVisIntake }