"use strict";

var _interopRequireDefault = require("@babel/runtime/helpers/interopRequireDefault");
Object.defineProperty(exports, "__esModule", {
  value: true
});
exports.AbstractGeoFileImporter = void 0;
var _defineProperty2 = _interopRequireDefault(require("@babel/runtime/helpers/defineProperty"));
var _i18n = require("@kbn/i18n");
var _public = require("@kbn/data-plugin/public");
var _importer = require("../importer");
var _constants = require("../../../common/constants");
var _geojson_clean_and_validate = require("./geojson_clean_and_validate");
var _create_chunks = require("./create_chunks");
/*
 * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
 * or more contributor license agreements. Licensed under the Elastic License
 * 2.0; you may not use this file except in compliance with the Elastic License
 * 2.0.
 */

const BLOCK_SIZE_MB = 5 * _constants.MB;
class AbstractGeoFileImporter extends _importer.Importer {
  constructor(file) {
    super();
    (0, _defineProperty2.default)(this, "_file", void 0);
    (0, _defineProperty2.default)(this, "_isActive", true);
    (0, _defineProperty2.default)(this, "_hasNext", true);
    (0, _defineProperty2.default)(this, "_features", []);
    (0, _defineProperty2.default)(this, "_totalBytesRead", 0);
    (0, _defineProperty2.default)(this, "_totalBytesImported", 0);
    (0, _defineProperty2.default)(this, "_blockSizeInBytes", 0);
    (0, _defineProperty2.default)(this, "_totalFeaturesRead", 0);
    (0, _defineProperty2.default)(this, "_totalFeaturesImported", 0);
    (0, _defineProperty2.default)(this, "_geometryTypesMap", new Map());
    (0, _defineProperty2.default)(this, "_invalidFeatures", []);
    (0, _defineProperty2.default)(this, "_geoFieldType", _public.ES_FIELD_TYPES.GEO_SHAPE);
    (0, _defineProperty2.default)(this, "_smallChunks", false);
    this._file = file;
  }
  destroy() {
    this._isActive = false;
  }
  canPreview() {
    return true;
  }
  renderEditor(onChange) {
    return null;
  }
  async previewFile(rowLimit, sizeLimit) {
    await this._readUntil(rowLimit, sizeLimit);
    return {
      features: [...this._features],
      invalidFeatures: [...this._invalidFeatures],
      previewCoverage: this._hasNext ? Math.round(this._getProgress(this._features.length, this._blockSizeInBytes)) : 100,
      hasPoints: this._geometryTypesMap.has('Point') || this._geometryTypesMap.has('MultiPoint'),
      hasShapes: this._geometryTypesMap.has('LineString') || this._geometryTypesMap.has('MultiLineString') || this._geometryTypesMap.has('Polygon') || this._geometryTypesMap.has('MultiPolygon') || this._geometryTypesMap.has('GeometryCollection')
    };
  }
  setGeoFieldType(geoFieldType) {
    this._geoFieldType = geoFieldType;
  }
  setSmallChunks(smallChunks) {
    this._smallChunks = smallChunks;
  }
  async import(id, index, pipelineId, setImportProgress) {
    if (!id || !index) {
      return {
        success: false,
        error: _i18n.i18n.translate('xpack.fileUpload.import.noIdOrIndexSuppliedErrorMessage', {
          defaultMessage: 'no ID or index supplied'
        })
      };
    }
    const maxChunkCharCount = this._smallChunks ? _importer.MAX_CHUNK_CHAR_COUNT / 10 : _importer.MAX_CHUNK_CHAR_COUNT;
    let success = true;
    const failures = [...this._invalidFeatures];
    let error;
    let importBlockPromise;

    // Read file in blocks to avoid loading too much of file into memory at a time
    while ((this._features.length > 0 || this._hasNext) && this._isActive) {
      await this._readUntil(undefined, BLOCK_SIZE_MB);
      if (!this._isActive) {
        return {
          success: false,
          failures
        };
      }

      // wait for previous import call to finish before starting next import
      if (importBlockPromise !== undefined) {
        const importBlockResults = await importBlockPromise;
        importBlockPromise = undefined;
        if (importBlockResults.failures) {
          failures.push(...importBlockResults.failures);
        }
        if (!importBlockResults.success) {
          success = false;
          error = importBlockResults.error;
          break;
        }
      }

      // Import block in chunks to avoid sending too much data to Elasticsearch at a time.
      const chunks = (0, _create_chunks.createChunks)(this._features, this._geoFieldType, maxChunkCharCount);
      const blockSizeInBytes = this._blockSizeInBytes;

      // reset block for next read
      this._features = [];
      this._blockSizeInBytes = 0;
      importBlockPromise = this._importBlock(id, index, pipelineId, chunks, blockSizeInBytes, setImportProgress);
    }

    // wait for last import call
    if (importBlockPromise) {
      const importBlockResults = await importBlockPromise;
      if (importBlockResults.failures) {
        failures.push(...importBlockResults.failures);
      }
      if (!importBlockResults.success) {
        success = false;
        error = importBlockResults.error;
      }
    }
    setImportProgress(100);
    return {
      success,
      failures,
      docCount: this._totalFeaturesRead,
      error
    };
  }
  async _importBlock(id, index, pipelineId, chunks, blockSizeInBytes, setImportProgress) {
    let success = true;
    const failures = [];
    let error;
    for (let i = 0; i < chunks.length; i++) {
      let retries = _importer.IMPORT_RETRIES;
      let resp = {
        success: false,
        failures: [],
        docCount: 0,
        id: '',
        index: '',
        pipelineId: ''
      };
      while (resp.success === false && retries > 0) {
        try {
          resp = await (0, _importer.callImportRoute)({
            id,
            index,
            data: chunks[i],
            settings: {},
            mappings: {},
            ingestPipeline: {
              id: pipelineId
            }
          });
          if (!this._isActive) {
            return {
              success: false,
              failures
            };
          }
          if (retries < _importer.IMPORT_RETRIES) {
            // eslint-disable-next-line no-console
            console.log(`Retrying import ${_importer.IMPORT_RETRIES - retries}`);
          }
          retries--;
        } catch (err) {
          resp.success = false;
          resp.error = err;
          retries = 0;
        }
      }
      if (resp.failures && resp.failures.length) {
        // failure.item is the document position in the chunk passed to import endpoint.
        // Need to update failure.item to reflect the actual feature position in the file.
        // e.g. item 3 in chunk is actually item 20003
        for (let f = 0; f < resp.failures.length; f++) {
          const failure = resp.failures[f];
          failure.item += this._totalFeaturesImported;
        }
        failures.push(...resp.failures);
      }
      if (resp.success) {
        this._totalFeaturesImported += chunks[i].length;

        // Advance block percentage in equal increments
        // even though chunks are not identical in size.
        // Reason being that chunk size does not exactly correlate to bytes read from file
        // because features are converted to elasticsearch documents which changes the size.
        const chunkProgress = (i + 1) / chunks.length;
        const totalBytesImported = this._totalBytesImported + blockSizeInBytes * chunkProgress;
        const importPercent = this._getProgress(this._totalFeaturesImported, totalBytesImported);
        setImportProgress(Math.round(importPercent * 10) / 10);
      } else {
        success = false;
        error = resp.error;
        break;
      }
    }
    this._totalBytesImported += blockSizeInBytes;
    return {
      success,
      failures,
      error
    };
  }
  async _readUntil(rowLimit, sizeLimit) {
    while (this._isActive && this._hasNext && (rowLimit === undefined || this._features.length < rowLimit) && (sizeLimit === undefined || this._blockSizeInBytes < sizeLimit)) {
      const results = await this._readNext(this._totalFeaturesRead, this._totalBytesRead);
      this._hasNext = results.hasNext;
      this._blockSizeInBytes = this._blockSizeInBytes + results.bytesRead;
      this._features = [...this._features, ...results.features.map(feature => {
        return (0, _geojson_clean_and_validate.geoJsonCleanAndValidate)(feature);
      })];
      results.geometryTypesMap.forEach((value, key) => {
        this._geometryTypesMap.set(key, value);
      });
      this._invalidFeatures = [...this._invalidFeatures, ...results.invalidFeatures];
      this._totalBytesRead = this._totalBytesRead + results.bytesRead;
      this._totalFeaturesRead = this._totalFeaturesRead + results.features.length + results.invalidFeatures.length;
    }
  }
  _readNext(prevFeaturesRead, prevBytesRead) {
    throw new Error('Should implement AbstractGeoFileImporter._next');
  }
  _getProgress(featuresProcessed, bytesProcessed) {
    return bytesProcessed / this._file.size * 100;
  }
  _getIsActive() {
    return this._isActive;
  }
  _getFile() {
    return this._file;
  }
  read(data) {
    throw new Error('read(data: ArrayBuffer) not supported, use previewFile and import instead.');
  }
  _createDocs(text) {
    throw new Error('_createDocs not implemented.');
  }
}
exports.AbstractGeoFileImporter = AbstractGeoFileImporter;