30084067687

Committed 24 Jul 2026 09:51AM UTC coverage: 94.176% (+0.06%) from 94.118%

Build # 30084067687

Build Type

Pull #247

github

Committed by

web-flow

Commit Message

Merge de80e38b5 into 04b9cfdbb

Pull Request Pull Request #247: Improve, add mongo cursor

Coverage Stats

1967 of 2453 branches covered (80.19%)

Branch coverage included in aggregate %.

227 of 228 new or added lines in 2 files covered. (99.56%)

23 existing lines in 2 files now uncovered.

7299 of 7386 relevant lines covered (98.82%)

196.99 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.23

/src/lib/fda.js

// Copyright 2025 Telefónica Soluciones de Informática y Comunicaciones de España, S.A.U.
// PROJECT: fiware-data-access
//
// This software and / or computer program has been developed by Telefónica Soluciones
// de Informática y Comunicaciones de España, S.A.U (hereinafter TSOL) and is protected
// as copyright by the applicable legislation on intellectual property.
//
// It belongs to TSOL, and / or its licensors, the exclusive rights of reproduction,
// distribution, public communication and transformation, and any economic right on it,
// all without prejudice of the moral rights of the authors mentioned above. It is expressly
// forbidden to decompile, disassemble, reverse engineer, sublicense or otherwise transmit
// by any means, translate or create derivative works of the software and / or computer
// programs, and perform with respect to all or part of such programs, any type of exploitation.
//
// Any use of all or part of the software and / or computer program will require the
// express written consent of TSOL. In all cases, it will be necessary to make
// an express reference to TSOL ownership in the software and / or computer
// program.
//
// Non-fulfillment of the provisions set forth herein and, in general, any violation of
// the peaceful possession and ownership of these rights will be prosecuted by the means
// provided in both Spanish and international law. TSOL reserves any civil or
// criminal actions it may exercise to protect its rights.

import { PassThrough } from 'node:stream';
import { getAgenda } from './jobs.js';
import {
  runPreparedStatement,
  runPreparedStatementStream,
  getDBConnection,
  releaseDBConnection,
  toParquet,
  copyQueryToParquet,
  checkParams,
  validateDAParamBindings,
  resolveDAParams,
  validateDAQuery,
  extractDate,
  PARTITION_TYPES,
  refreshIntervalPartitionCheck,
} from './utils/db.js';
import {
  uploadTable,
  runPgQuery,
  createPgCursorReader,
  validatePostgresDatasourceConnection,
  validatePostgresQuery,
} from './utils/pg.js';
import {
  getS3Client,
  newUpload,
  dropFile,
  moveObject,
  listObjects,
  dropFiles,
} from './utils/aws.js';
import {
  createFDAMongo,
  regenerateFDA,
  retrieveFDAs,
  retrieveFDA,
  storeDA,
  removeFDA,
  retrieveDAs,
  retrieveDA,
  updateDA,
  removeDA,
  updateFDAStatus,
  createDatasource,
  retrieveDatasources,
  retrieveDatasource,
  updateDatasource,
  removeDatasource,
  countFDAsUsingDatasource,
  validateMongoDatasourceConnection,
  createMongoCursorReader,
} from './utils/mongo.js';
import {
  normalizeForSerialization,
  getWindowDate,
  assertFreshQueriesEnabled,
  acquireFreshQuerySlot,
  convertRefreshIntervalToMs,
  processFetchSize,
} from './utils/utils.js';
import {
  buildFDAJobFilter,
  buildFDAJobCancelFilter,
  getBucketNameFromService,
  getFDAStoragePath,
  normalizeServicePath,
} from './utils/fdaScope.js';
import { config } from './fdaConfig.js';
import { FDAError } from './fdaError.js';

const FDA_VALIDATION_MODE_STRICT = 'strict';
const FDA_VALIDATION_MODE_UNCHECKED = 'unchecked';

const FRESH_CURSOR_BATCH_SIZE = 250;

const DEFAULT_DATASOURCE_ID = 'default';
const SUPPORTED_DATASOURCE_TYPES = new Set(['postgres', 'mongodb']);

function assertSupportedDatasourceType(type) {
  if (!SUPPORTED_DATASOURCE_TYPES.has(type)) {
    throw new FDAError(
      400,
      'UnsupportedDatasourceType',
      `Datasource type ${type} is not supported for this operation`,
    );
  }
}

export function validateMongoFDAContract(query, timeColumn, cached) {
  validateBasicQueryStructure(query);

  const { collection, filter, projection, aggregation } = query;
  validateCollection(collection);

  const queryType = determineQueryType(filter, aggregation);

  if (queryType === 'find') {
    validateFindQuery(filter, projection, timeColumn);
  } else if (queryType === 'aggregation') {
    validateAggregationQuery(aggregation);
  }

  validateCacheSupport(cached);
}

// Helper functions to reduce complexity
function validateBasicQueryStructure(query) {
  if (!query || typeof query !== 'object' || Array.isArray(query)) {
    throw new FDAError(
      400,
      'InvalidMongoFDAContract',
      'Mongo FDA query must be a JSON object',
    );
  }
}

function validateCollection(collection) {
  if (!collection || typeof collection !== 'string') {
    throw new FDAError(
      400,
      'InvalidMongoFDAContract',
      'Mongo FDA query requires a non-empty collection field',
    );
  }
}

function determineQueryType(filter, aggregation) {
  const hasFindQuery = filter !== undefined;
  const hasAggregationQuery = aggregation !== undefined;

  if (hasFindQuery === hasAggregationQuery) {
    throw new FDAError(
      400,
      'InvalidMongoFDAContract',
      'Mongo FDA query must define either filter or aggregation',
    );
  }

  return hasFindQuery ? 'find' : 'aggregation';
}

function validateFindQuery(filter, projection, timeColumn) {
  validateFilter(filter);
  validateProjection(projection);
  validateTimeColumnInProjection(timeColumn, projection);
}

function validateFilter(filter) {
  if (filter === null || typeof filter !== 'object' || Array.isArray(filter)) {
    throw new FDAError(
      400,
      'InvalidMongoFDAContract',
      'Mongo FDA filter must be a JSON object',
    );
  }
}

function validateProjection(projection) {
  if (
    projection !== undefined &&
    (projection === null ||
      typeof projection !== 'object' ||
      Array.isArray(projection))
  ) {
    throw new FDAError(
      400,
      'InvalidMongoFDAContract',
      'Mongo FDA projection must be an object',
    );
  }
}

function validateTimeColumnInProjection(timeColumn, projection) {
  if (timeColumn && projection && !(timeColumn in projection)) {
    throw new FDAError(
      400,
      'InvalidMongoFDAContract',
      'Mongo FDA timeColumn must be included in projection',
    );
  }
}

function validateAggregationQuery(aggregation) {
  if (!Array.isArray(aggregation) || aggregation.length === 0) {
    throw new FDAError(
      400,
      'InvalidMongoFDAContract',
      'Mongo FDA aggregation must be a non-empty array',
    );
  }
  throw new FDAError(
    400,
    'MongoAggregationNotSupported',
    'Aggregation pipelines are not supported yet',
  );
}

function validateCacheSupport(cached) {
  if (cached === false) {
    throw new FDAError(
      400,
      'InvalidMongoFDAContract',
      'Mongo datasource only supports cached FDAs',
    );
  }
}

async function resolveDatasource(service, datasourceId) {
  const ds = await retrieveDatasource(service, datasourceId);
  if (!ds) {
    throw new FDAError(
      404,
      'DatasourceNotFound',
      `Datasource ${datasourceId} not found for service ${service}`,
    );
  }

  assertSupportedDatasourceType(ds.type);
  return ds;
}

async function resolveDatasourceCredentials(service, datasourceId) {
  const ds = await resolveDatasource(service, datasourceId);
  return ds.config;
}

async function validateDatasourceConnection(type, dsConfig) {
  assertSupportedDatasourceType(type);

  if (type === 'postgres') {
    await validatePostgresDatasourceConnection(dsConfig);
  } else {
    await validateMongoDatasourceConnection(dsConfig);
  }
}

export async function createDatasourceForService(
  service,
  datasourceId,
  type,
  dsConfig,
) {
  await validateDatasourceConnection(type, dsConfig);
  await createDatasource(service, datasourceId, type, dsConfig);
}

export function getDatasourcesForService(service) {
  return retrieveDatasources(service);
}

export async function getDatasourceForService(service, datasourceId) {
  const ds = await retrieveDatasource(service, datasourceId);
  if (!ds) {
    throw new FDAError(
      404,
      'DatasourceNotFound',
      `Datasource ${datasourceId} not found for service ${service}`,
    );
  }
  return ds;
}

export async function updateDatasourceForService(
  service,
  datasourceId,
  type,
  dsConfig,
) {
  if (type !== undefined || dsConfig !== undefined) {
    const current = await getDatasourceForService(service, datasourceId);
    await validateDatasourceConnection(
      type ?? current.type,
      dsConfig ?? current.config,
    );
  }

  await updateDatasource(service, datasourceId, type, dsConfig);
}

export async function deleteDatasourceForService(service, datasourceId) {
  const usedBy = await countFDAsUsingDatasource(service, datasourceId);
  if (usedBy > 0) {
    throw new FDAError(
      409,
      'DatasourceInUse',
      `Datasource ${datasourceId} is being used by ${usedBy} FDA(s) in service ${service}`,
    );
  }

  await removeDatasource(service, datasourceId);
}
export const VALID_VISIBILITIES = ['public', 'private'];
const VALID_VISIBILITIES_SET = new Set(VALID_VISIBILITIES);
const VALID_REFRESH_POLICY_TYPES = ['none', 'interval', 'window'];
const VALID_WINDOW_FETCH_SIZES = ['hour', 'day', 'week', 'month', 'year'];
const CSV_CONTENT_TYPE = 'text/csv; charset=utf-8';

function stringifyCsvValue(value) {
  const normalizedValue = normalizeForSerialization(value);

  if (normalizedValue === null || normalizedValue === undefined) {
    return '';
  }

  if (typeof normalizedValue === 'object') {
    return JSON.stringify(normalizedValue);
  }

  return String(normalizedValue);
}

function escapeCsvValue(value) {
  const strValue = stringifyCsvValue(value);

  if (
    strValue.includes(',') ||
    strValue.includes('"') ||
    strValue.includes('\n') ||
    strValue.includes('\r')
  ) {
    return '"' + strValue.replace(/"/g, '""') + '"';
  }

  return strValue;
}

async function writeCsvLine(res, line) {
  const ok = res.write(line);
  if (!ok) {
    await new Promise((resolve) => res.once('drain', resolve));
  }
}

async function writeNdjsonLine(res, row) {
  const safeObj = normalizeForSerialization(row);
  const ok = res.write(JSON.stringify(safeObj) + '\n');
  if (!ok) {
    await new Promise((resolve) => res.once('drain', resolve));
  }
}

async function writeCsvHeader(res, columnNames) {
  if (columnNames.length === 0) {
    return;
  }

  await writeCsvLine(
    res,
    columnNames.map((columnName) => escapeCsvValue(columnName)).join(',') +
      '\n',
  );
}

function toRowObject(row, columnNames) {
  const rowObj = {};

  for (let i = 0; i < columnNames.length; i++) {
    rowObj[columnNames[i]] = row[i];
  }

  return rowObj;
}

export async function getFDAs(service, visibility, servicePath) {
  const fdas = await retrieveFDAs(service);
  const normalizedServicePath = normalizeServicePath(servicePath);

  if (visibility === undefined) {
    return fdas
      .filter(
        (fda) =>
          normalizeServicePath(fda.servicePath) === normalizedServicePath,
      )
      .map((fda) => toFDAApiResponse(fda, { includeId: true }));
  }

  const normalizedVisibility = normalizeVisibility(visibility);

  return fdas
    .filter(
      (fda) =>
        normalizeVisibility(fda.visibility) === normalizedVisibility &&
        normalizeServicePath(fda.servicePath) === normalizedServicePath,
    )
    .map((fda) => toFDAApiResponse(fda, { includeId: true }));
}

export async function getFDA(service, fdaId, visibility, servicePath) {
  const normalizedServicePath = normalizeServicePath(servicePath);

  if (visibility === undefined) {
    const fda = await getStoredFDA(service, fdaId, normalizedServicePath);
    return toFDAApiResponse(fda, { includeId: false });
  }

  const fda = await getAccessibleFDA(service, fdaId, visibility, servicePath);
  return toFDAApiResponse(fda, { includeId: false });
}

export async function executeQuery({
  service,
  visibility,
  servicePath,
  params,
  fresh = false,
}) {
  if (fresh) {
    return executeFreshQuery({ service, visibility, servicePath, params });
  }

  const { fdaId, daId, ...rest } = params;

  await ensureFDAReadyForQuery(service, fdaId, visibility, servicePath);

  const conn = await getDBConnection();

  try {
    const rows = await runPreparedStatement(
      conn,
      service,
      fdaId,
      daId,
      rest,
      servicePath,
    );

    return normalizeForSerialization(rows);
  } finally {
    await releaseDBConnection(conn);
  }
}

export async function executeFDAQuery({
  service,
  visibility,
  servicePath,
  fdaId,
}) {
  assertFreshQueriesEnabled(config.roles.syncQueries);

  const releaseFreshSlot = acquireFreshQuerySlot(
    config.freshQueries.maxConcurrent,
  );
  try {
    const fda = await getAccessibleFDA(service, fdaId, visibility, servicePath);
    const query = buildFreshQueryFromFDA(fda, fdaId);
    const pgCredentials = await resolveDatasourceCredentials(
      service,
      fda.datasourceId ?? DEFAULT_DATASOURCE_ID,
    );
    const rows = await runPgQuery(pgCredentials, query, []);
    return normalizeForSerialization(rows);
  } finally {
    releaseFreshSlot();
  }
}

export async function executeQueryStream({
  service,
  visibility,
  servicePath,
  params,
  req,
  res,
  fresh = false,
  format = 'ndjson',
}) {
  const source = fresh
    ? await createFreshRowSource({
        service,
        visibility,
        servicePath,
        params,
        req,
      })
    : await createCachedRowSource({
        service,
        visibility,
        servicePath,
        params,
        req,
      });

  if (format === 'ndjson') {
    res.setHeader('Content-Type', 'application/x-ndjson');
  } else {
    res.setHeader('Content-Type', CSV_CONTENT_TYPE);
    res.setHeader('Content-Disposition', 'attachment; filename="results.csv"');
  }

  let csvColumns;

  try {
    for (
      let rows = await source.readNextRows();
      rows.length > 0;
      rows = await source.readNextRows()
    ) {
      if (format === 'ndjson') {
        for (const row of rows) {
          const rowObj = source.columnNames
            ? toRowObject(row, source.columnNames)
            : row;
          await writeNdjsonLine(res, rowObj);
        }
        continue;
      }

      if (!csvColumns) {
        csvColumns = source.columnNames ?? Object.keys(rows[0] ?? {});
        await writeCsvHeader(res, csvColumns);
      }

      for (const row of rows) {
        const csvLine = source.columnNames
          ? row.map((cell) => escapeCsvValue(cell)).join(',') + '\n'
          : csvColumns.map((column) => escapeCsvValue(row[column])).join(',') +
            '\n';
        await writeCsvLine(res, csvLine);
      }
    }
  } finally {
    await source.close();
  }

  return res.end();
}

export async function executeFDAQueryStream({
  service,
  visibility,
  servicePath,
  fdaId,
  req,
  res,
  format = 'ndjson',
}) {
  const source = await createFreshFDARowSource({
    service,
    visibility,
    servicePath,
    fdaId,
    req,
  });

  if (format === 'ndjson') {
    res.setHeader('Content-Type', 'application/x-ndjson');
  } else {
    res.setHeader('Content-Type', CSV_CONTENT_TYPE);
    res.setHeader('Content-Disposition', 'attachment; filename="results.csv"');
  }

  let csvColumns;

  try {
    for (
      let rows = await source.readNextRows();
      rows.length > 0;
      rows = await source.readNextRows()
    ) {
      if (format === 'ndjson') {
        for (const row of rows) {
          await writeNdjsonLine(res, row);
        }
        continue;
      }

      if (!csvColumns) {
        csvColumns = Object.keys(rows[0] ?? {});
        await writeCsvHeader(res, csvColumns);
      }

      for (const row of rows) {
        const csvLine =
          csvColumns.map((column) => escapeCsvValue(row[column])).join(',') +
          '\n';
        await writeCsvLine(res, csvLine);
      }
    }
  } finally {
    await source.close();
  }

  return res.end();
}

async function createCachedRowSource({
  service,
  visibility,
  servicePath,
  params,
  req,
}) {
  const { fdaId, daId, ...rest } = params;

  await ensureFDAReadyForQuery(service, fdaId, visibility, servicePath);

  const conn = await getDBConnection();

  let stream;
  let closeStream;

  try {
    const result = await runPreparedStatementStream(
      conn,
      service,
      fdaId,
      daId,
      rest,
      servicePath,
    );

    stream = result.stream;
    closeStream = result.close;
  } catch (err) {
    await releaseDBConnection(conn);
    throw err;
  }

  let cleaned = false;
  const cleanup = async () => {
    if (cleaned) {
      return;
    }
    cleaned = true;

    try {
      await closeStream();
    } finally {
      await releaseDBConnection(conn);
    }
  };

  req.on('close', () => {
    cleanup().catch(() => {});
  });

  return {
    columnNames: stream.columnNames(),
    async readNextRows() {
      const chunk = await stream.fetchChunk();
      return chunk.rowCount > 0 ? chunk.getRows() : [];
    },
    close: cleanup,
  };
}

async function createFreshRowSource({
  service,
  visibility,
  servicePath,
  params,
  req,
}) {
  assertFreshQueriesEnabled(config.roles.syncQueries);

  const releaseFreshSlot = acquireFreshQuerySlot(
    config.freshQueries.maxConcurrent,
  );
  let cursorReader;

  try {
    const { text, values, fda } = await buildFreshQueryStatement(
      service,
      visibility,
      servicePath,
      params,
    );
    const pgCredentials = await resolveDatasourceCredentials(
      service,
      fda.datasourceId ?? DEFAULT_DATASOURCE_ID,
    );

    cursorReader = await createPgCursorReader(
      pgCredentials,
      text,
      values,
      FRESH_CURSOR_BATCH_SIZE,
    );

    req.on('close', () => {
      cursorReader?.close().catch(() => {});
    });

    return {
      columnNames: null,
      readNextRows: () => cursorReader.readNextChunk(),
      close: async () => {
        await cursorReader?.close();
        releaseFreshSlot();
      },
    };
  } catch (e) {
    releaseFreshSlot();
    throw e;
  }
}

async function createFreshFDARowSource({
  service,
  visibility,
  servicePath,
  fdaId,
  req,
}) {
  assertFreshQueriesEnabled(config.roles.syncQueries);

  const releaseFreshSlot = acquireFreshQuerySlot(
    config.freshQueries.maxConcurrent,
  );
  let cursorReader;

  try {
    const { query, fda } = await buildFreshFDAQuery(
      service,
      visibility,
      servicePath,
      fdaId,
    );
    const pgCredentials = await resolveDatasourceCredentials(
      service,
      fda.datasourceId ?? DEFAULT_DATASOURCE_ID,
    );

    cursorReader = await createPgCursorReader(
      pgCredentials,
      query,
      [],
      FRESH_CURSOR_BATCH_SIZE,
    );

    req.on('close', () => {
      cursorReader?.close().catch(() => {});
    });

    return {
      columnNames: null,
      readNextRows: () => cursorReader.readNextChunk(),
      close: async () => {
        await cursorReader?.close();
        releaseFreshSlot();
      },
    };
  } catch (e) {
    releaseFreshSlot();
    throw e;
  }
}

async function executeFreshQuery({ service, visibility, servicePath, params }) {
  assertFreshQueriesEnabled(config.roles.syncQueries);

  const releaseFreshSlot = acquireFreshQuerySlot(
    config.freshQueries.maxConcurrent,
  );
  try {
    const { text, values, fda } = await buildFreshQueryStatement(
      service,
      visibility,
      servicePath,
      params,
    );
    const pgCredentials = await resolveDatasourceCredentials(
      service,
      fda.datasourceId ?? DEFAULT_DATASOURCE_ID,
    );
    const rows = await runPgQuery(pgCredentials, text, values);
    return normalizeForSerialization(rows);
  } catch (e) {
    if (e instanceof FDAError) {
      throw e;
    }

    throw e;
  } finally {
    releaseFreshSlot();
  }
}

async function buildFreshQueryStatement(
  service,
  visibility,
  servicePath,
  params,
) {
  const { fdaId, daId, ...rest } = params;

  const da = await retrieveDA(service, fdaId, daId, servicePath);
  if (!da?.query) {
    throw new FDAError(
      404,
      'DaNotFound',
      `DA ${daId} does not exist in FDA ${fdaId} with service ${service}.`,
    );
  }

  const fda = await getAccessibleFDA(service, fdaId, visibility, servicePath);

  const validatedParams = resolveDAParams(rest || {}, da.params);
  const freshBaseQuery = buildFreshDAQuery(fda.query, da.query);

  return {
    ...replaceNamedParamsWithPositional(freshBaseQuery, validatedParams),
    fda,
  };
}

async function buildFreshFDAQuery(service, visibility, servicePath, fdaId) {
  const fda = await getAccessibleFDA(service, fdaId, visibility, servicePath);
  return { query: buildFreshQueryFromFDA(fda, fdaId), fda };
}

function buildFreshQueryFromFDA(fda, fdaId) {
  if (fda.cached !== false) {
    throw new FDAError(
      409,
      'FDANotOnlyFresh',
      `FDA ${fdaId} is a cached FDA and cannot be queried directly. Use a Data Access instead.`,
    );
  }
  return removeTrailingSemicolon(fda.query?.trim() || '');
}

function buildFreshDAQuery(fdaQuery, daQuery) {
  const cleanFdaQuery = removeTrailingSemicolon(fdaQuery?.trim() || '');
  const cleanDaQuery = removeTrailingSemicolon(daQuery?.trim() || '');

  if (!cleanDaQuery || /^from\b/i.test(cleanDaQuery)) {
    throw new FDAError(
      400,
      'InvalidDAQuery',
      'DA query must not include FROM clause at start. It is managed internally.',
    );
  }

  if (!/^select\b/i.test(cleanDaQuery)) {
    throw new FDAError(
      400,
      'InvalidDAQuery',
      'Fresh query mode requires DA query to start with SELECT.',
    );
  }

  const selectTail = cleanDaQuery.replace(/^select\s+/i, '');
  const clauseMatch = selectTail.match(
    /\b(where|group\s+by|having|order\s+by|limit|offset)\b/i,
  );

  const projection = clauseMatch
    ? selectTail.slice(0, clauseMatch.index).trim()
    : selectTail.trim();
  const clauses = clauseMatch ? selectTail.slice(clauseMatch.index).trim() : '';

  if (!projection) {
    throw new FDAError(
      400,
      'InvalidDAQuery',
      'DA query must contain a SELECT projection.',
    );
  }

  if (/\bfrom\b/i.test(projection)) {
    throw new FDAError(
      400,
      'InvalidDAQuery',
      'DA query must not include FROM clause. It is managed internally.',
    );
  }

  const trailing = clauses ? ` ${clauses}` : '';
  return `SELECT ${projection} FROM (${cleanFdaQuery}) AS fda_source${trailing}`;
}

function replaceNamedParamsWithPositional(query, params) {
  const indexes = new Map();
  const values = [];

  const text = query.replaceAll(/\$([A-Za-z_]\w*)/g, (_m, name) => {
    if (!Object.prototype.hasOwnProperty.call(params, name)) {
      /* c8 ignore next 5 */
      throw new FDAError(
        400,
        'InvalidQueryParam',
        `Missing required param "${name}".`,
      );
    }

    if (!indexes.has(name)) {
      indexes.set(name, values.length + 1);
      values.push(params[name]);
    }

    return `$${indexes.get(name)}`;
  });

  return { text, values };
}

function removeTrailingSemicolon(query) {
  return query.replace(/;+\s*$/, '');
}

export async function createDA(
  service,
  fdaId,
  daId,
  description,
  userQuery,
  params,
  visibility,
  servicePath,
) {
  const conn = await getDBConnection();

  try {
    let fda;
    if (visibility !== undefined || servicePath !== undefined) {
      fda = await getAccessibleFDA(service, fdaId, visibility, servicePath);
      assertFDAIsCached(fda, fdaId);
    }

    if (!fda) {
      fda = await retrieveFDA(service, fdaId, servicePath);
    }

    const existing = await retrieveDA(service, fdaId, daId, servicePath);

    if (existing) {
      throw new FDAError(
        409,
        'DuplicatedKey',
        `DA ${daId} already exists in FDA ${fdaId}`,
      );
    }

    const normalizedParams = checkParams(params);
    validateDAParamBindings(userQuery, normalizedParams || []);
    if (shouldValidateDACompatibility(fda)) {
      await validateDAQuery(conn, service, fdaId, userQuery, servicePath);
    }
    await storeDA(
      service,
      fdaId,
      servicePath,
      daId,
      description,
      userQuery,
      normalizedParams,
    );
  } finally {
    await releaseDBConnection(conn);
  }
}

export async function fetchFDA(
  fdaId,
  query,
  service,
  visibility,
  servicePath,
  description,
  refreshPolicy,
  timeColumn,
  objStgConf,
  defaultDataAccessEnabled,
  cached = true,
  datasourceId = DEFAULT_DATASOURCE_ID,
  validationMode = FDA_VALIDATION_MODE_STRICT,
) {
  const normalizedVisibility = normalizeVisibility(visibility);
  const normalizedServicePath = normalizeServicePath(servicePath);
  const datasource = await getDatasourceForService(service, datasourceId);
  validateScheduledOptions(refreshPolicy, objStgConf, timeColumn);

  if (datasource.type === 'mongodb') {
    validateMongoFDAContract(query, timeColumn, cached);

    if (refreshPolicy?.type === 'window') {
      throw new FDAError(
        400,
        'InvalidMongoFDAContract',
        'Mongo datasource does not support window refresh policy',
      );
    }
  }

  const sourceSchema = await validateAndGetSourceSchema(
    datasource,
    validationMode,
    query,
    timeColumn,
  );

  const persistedSchema = buildPersistedSchema(sourceSchema);

  await createFDAMongo(
    fdaId,
    query,
    service,
    normalizedVisibility,
    normalizedServicePath,
    description,
    refreshPolicy,
    timeColumn,
    objStgConf,
    cached,
    datasourceId,
    validationMode,
    persistedSchema,
  );

  if (cached && validationMode === FDA_VALIDATION_MODE_STRICT) {
    await prepareCachedFDA({
      service,
      fdaId,
      query,
      servicePath: normalizedServicePath,
      datasourceId,
      timeColumn,
      objStgConf,
      defaultDataAccessEnabled,
      visibility: normalizedVisibility,
      sourceSchema,
    });
  }

  if (!cached) {
    return;
  }

  // Schedule refreshes according to the refresh policy
  await scheduleFDAJobs({
    fdaId,
    query,
    service,
    servicePath: normalizedServicePath,
    timeColumn,
    refreshPolicy,
    objStgConf,
    datasource,
  });
}

function validateAndGetSourceSchema(
  datasource,
  validationMode,
  query,
  timeColumn,
) {
  if (
    datasource.type !== 'postgres' ||
    validationMode !== FDA_VALIDATION_MODE_STRICT
  ) {
    return null;
  }

  return validatePostgresQuery(datasource.config, query, {
    timeColumn,
    returnColumns: true,
  });
}

async function prepareCachedFDA({
  service,
  fdaId,
  query,
  servicePath,
  datasourceId,
  timeColumn,
  objStgConf,
  defaultDataAccessEnabled,
  visibility,
  sourceSchema,
}) {
  try {
    await createParquet(
      service,
      fdaId,
      query,
      servicePath,
      datasourceId,
      timeColumn,
      objStgConf,
      sourceSchema,
    );
  } catch (err) {
    await rollbackFDAProvisioning(service, fdaId, servicePath);
    throw err;
  }

  await createDefaultDAIfNeeded({
    cached: true,
    defaultDataAccessEnabled,
    service,
    fdaId,
    normalizedServicePath: servicePath,
    timeColumn,
    objStgConf,
    normalizedVisibility: visibility,
    sourceSchema,
  });
}

function resolveRefreshQueries(datasource, refreshPolicy, query, timeColumn) {
  if (refreshPolicy?.type !== 'window') {
    return {
      firstQuery: query,
      recurringQuery: query,
    };
  }

  if (datasource.type === 'mongodb') {
    throw new FDAError(
      400,
      'InvalidMongoFDAContract',
      'Mongo datasource does not support window refresh policy',
    );
  }

  return {
    firstQuery: getWindowQuery(
      query,
      timeColumn,
      refreshPolicy.params.windowSize,
    ),
    recurringQuery: getWindowQuery(
      query,
      timeColumn,
      refreshPolicy.params.fetchSize,
    ),
  };
}

function getWindowQuery(query, timeColumn, startDate) {
  if (!startDate) {
    return query;
  }
  const prevWindowStartDate = getPreviousWindowStartDate(startDate);
  return getUpdateWindowQuery(query, timeColumn, prevWindowStartDate);
}

function validateScheduledOptions(refreshPolicy, objStgConf, timeColumn) {
  if (!refreshPolicy) {
    return;
  }

  if (!VALID_REFRESH_POLICY_TYPES.includes(refreshPolicy.type)) {
    throw new FDAError(
      400,
      'InvalidParam',
      `Invalid refresh policy type "${refreshPolicy.type}".`,
    );
  }

  if (refreshPolicy.type === 'none') {
    return;
  }

  if (
    (refreshPolicy?.type === 'window' || objStgConf?.partition) &&
    !timeColumn
  ) {
    throw new FDAError(
      400,
      'InvalidParam',
      'timeColumn is required when using window refresh policy or partitioning.',
    );
  }

  const { refreshInterval, fetchSize } = refreshPolicy.params || {};
  const consistencyRefreshInterval =
    refreshPolicy.params?.consistencyRefreshInterval;
  if (!refreshInterval) {
    throw new FDAError(
      400,
      'InvalidParam',
      `Missing required refresh policy parameter: refreshInterval.`,
    );
  }

  if (convertRefreshIntervalToMs(refreshInterval) === null) {
    throw new FDAError(
      400,
      'InvalidParam',
      `Invalid refresh interval "${refreshInterval}".`,
    );
  }

  if (refreshPolicy.type === 'window' && !fetchSize) {
    throw new FDAError(
      400,
      'InvalidParam',
      `Missing required refresh policy parameter: fetchSize.`,
    );
  }

  if (refreshPolicy.type === 'window' && fetchSize) {
    const { unit } = processFetchSize(fetchSize);
    if (!VALID_WINDOW_FETCH_SIZES.includes(unit)) {
      throw new FDAError(
        400,
        'InvalidParam',
        `Invalid fetchSize "${fetchSize}".`,
      );
    }
  }

  if (
    refreshPolicy.type === 'window' &&
    refreshPolicy.params?.windowSize &&
    !objStgConf?.partition
  ) {
    throw new FDAError(
      400,
      'InvalidParam',
      'windowSize requires objStgConf.partition.',
    );
  }

  if (refreshPolicy.params?.windowSize) {
    const { unit } = processFetchSize(refreshPolicy.params.windowSize);
    if (!getWindowDate(unit)) {
      throw new FDAError(
        400,
        'InvalidParam',
        `Invalid windowSize "${refreshPolicy.params.windowSize}".`,
      );
    }
  }

  if (
    objStgConf?.partition &&
    !PARTITION_TYPES.includes(objStgConf.partition)
  ) {
    throw new FDAError(
      400,
      'InvalidParam',
      `Invalid partition type "${objStgConf.partition}".`,
    );
  }

  if (consistencyRefreshInterval) {
    if (refreshPolicy.type !== 'window') {
      throw new FDAError(
        400,
        'InvalidParam',
        'consistencyRefreshInterval is only supported for window refresh policy.',
      );
    }

    if (convertRefreshIntervalToMs(consistencyRefreshInterval) === null) {
      throw new FDAError(
        400,
        'InvalidParam',
        `Invalid consistency refresh interval "${consistencyRefreshInterval}".`,
      );
    }

    const refreshMs = convertRefreshIntervalToMs(refreshInterval);
    const consistencyMs = convertRefreshIntervalToMs(
      consistencyRefreshInterval,
    );

    if (
      refreshMs !== null &&
      consistencyMs !== null &&
      consistencyMs < refreshMs
    ) {
      throw new FDAError(
        400,
        'InvalidParam',
        `consistencyRefreshInterval ("${consistencyRefreshInterval}") must be greater than refreshInterval ("${refreshInterval}").`,
      );
    }
  }

  // RefreshInterval must be smaller or equal than partition size
  if (!refreshIntervalPartitionCheck(refreshInterval, objStgConf?.partition)) {
    throw new FDAError(
      400,
      'InvalidParam',
      `Refresh interval "${refreshInterval}" must be smaller or equal than partition size "${objStgConf?.partition}".`,
    );
  }

  // fetched data size must be equal than partition size (if both presents).
  if (objStgConf?.partition && fetchSize !== objStgConf?.partition) {
    throw new FDAError(
      400,
      'InvalidParam',
      `Fetch size "${fetchSize}" must be equal to partition size "${objStgConf?.partition}".`,
    );
  }
}

export async function updateFDA(service, fdaId, visibility, servicePath) {
  const normalizedServicePath = normalizeServicePath(servicePath);
  const fda =
    visibility !== undefined
      ? await getAccessibleFDA(service, fdaId, visibility, servicePath)
      : await getStoredFDA(service, fdaId, normalizedServicePath);

  assertFDAIsCached(fda, fdaId);

  const previous = await regenerateFDA(service, fdaId, normalizedServicePath);

  const agenda = getAgenda();

  // Execute refresh immediately (when a fetcher is free)
  const effectiveServicePath = previous.servicePath ?? normalizedServicePath;

  let firstQuery = previous.query;
  if (previous.refreshPolicy?.type === 'window') {
    firstQuery = getWindowQuery(
      previous.query,
      previous.timeColumn,
      previous.refreshPolicy?.params?.windowSize,
    );
  }
  await agenda.now('refresh-fda', {
    fdaId,
    query: firstQuery,
    service,
    servicePath: effectiveServicePath,
    timeColumn: previous.timeColumn,
    refreshPolicy: previous.refreshPolicy,
    objStgConf: previous.objStgConf,
    datasourceId: previous.datasourceId ?? DEFAULT_DATASOURCE_ID,
  });

  if (previous.refreshPolicy?.params?.windowSize) {
    await agenda.now('clean-partition', {
      fdaId,
      service,
      servicePath: effectiveServicePath,
      windowSize: previous.refreshPolicy.params.windowSize,
      objStgConf: previous.objStgConf,
    });
  }
}

export async function processFDAAsync(
  fdaId,
  query,
  service,
  servicePath,
  timeColumn,
  refreshPolicy,
  objStgConf,
  datasourceId = DEFAULT_DATASOURCE_ID,
) {
  const datasource = await resolveDatasource(service, datasourceId);

  if (datasource.type === 'mongodb' && refreshPolicy?.type === 'window') {
    throw new FDAError(
      400,
      'InvalidMongoFDAContract',
      'Mongo datasource does not support window refresh',
    );
  }

  const storagePath = getFDAStoragePath(fdaId, servicePath);
  const bucketName = getBucketNameFromService(service);

  try {
    await updateFDAStatus({
      service,
      fdaId,
      servicePath,
      status: 'fetching',
      progress: 10,
    });

    await uploadTableToObjStg(
      service,
      datasourceId,
      query,
      bucketName,
      storagePath,
      fdaId,
      servicePath,
      timeColumn,
      objStgConf,
    );

    await updateFDAStatus({
      service,
      fdaId,
      servicePath,
      status: 'completed',
      progress: 100,
    });
  } catch (err) {
    await updateFDAStatus({
      service,
      fdaId,
      servicePath,
      status: 'failed',
      progress: 0,
      error: err.message,
    });
    throw err;
  }
}

function getPreviousWindowStartDate(fetchSize) {
  const now = new Date();
  const { amount, unit } = processFetchSize(fetchSize);

  switch (unit) {
    case 'hour': {
      const d = new Date(now);
      d.setUTCHours(d.getUTCHours() - amount, 0, 0, 0);
      return d.toISOString();
    }

    case 'day': {
      const d = new Date(now);
      d.setUTCDate(d.getUTCDate() - amount);
      d.setUTCHours(0, 0, 0, 0);
      return d.toISOString();
    }

    case 'week': {
      const d = new Date(now);
      d.setUTCDate(d.getUTCDate() - 7 * amount);
      d.setUTCHours(0, 0, 0, 0);
      return d.toISOString();
    }

    case 'month': {
      const d = new Date(now);
      d.setUTCMonth(d.getUTCMonth() - amount);
      d.setUTCHours(0, 0, 0, 0);
      return d.toISOString();
    }

    case 'year': {
      const d = new Date(now);
      d.setUTCFullYear(d.getUTCFullYear() - amount);
      d.setUTCHours(0, 0, 0, 0);
      return d.toISOString();
    }

    default:
      throw new FDAError(
        400,
        'InvalidParam',
        `Invalid unit in window param: ${fetchSize}.`,
      );
  }
}

function getUpdateWindowQuery(query, timeColumn, latestFetchStartDate) {
  return `SELECT * FROM (${query}) q WHERE ${timeColumn} >= TIMESTAMP '${latestFetchStartDate}' AND ${timeColumn} < NOW()`;
}

function buildPersistedSchema(sourceSchema) {
  const schemaFields = Array.isArray(sourceSchema?.fields)
    ? sourceSchema.fields.filter(
        (field) =>
          typeof field?.name === 'string' &&
          field.name.length > 0 &&
          typeof field?.duckdbType === 'string' &&
          field.duckdbType.length > 0,
      )
    : [];

  if (schemaFields.length === 0) {
    return null;
  }

  return schemaFields.map(({ name, duckdbType }) => ({
    name,
    type: duckdbType,
  }));
}

function shouldValidateDACompatibility(fda) {
  return (
    (fda?.validationMode ?? FDA_VALIDATION_MODE_STRICT) !==
    FDA_VALIDATION_MODE_UNCHECKED
  );
}

async function uploadMongoCursorContentToObjectStorage(
  s3Client,
  bucket,
  path,
  reader,
) {
  const uploadBody = new PassThrough();
  const upload = newUpload(s3Client, bucket, `${path}.csv`, uploadBody, 5, 1);

  const uploadDone = upload.done();

  try {
    const columns = reader.columns || [];
    let wroteHeader = false;

    let rows = await reader.readNextChunk();
    while (rows.length > 0) {
      if (!wroteHeader) {
        if (columns.length > 0) {
          await writeCsvHeader(uploadBody, columns);
        }
        wroteHeader = true;
      }

      for (const row of rows) {
        const csvLine = columns
          .map((column) => escapeCsvValue(row[column]))
          .join(',');
        await writeCsvLine(uploadBody, `${csvLine}\n`);
      }

      rows = await reader.readNextChunk();
    }

    if (!wroteHeader && columns.length > 0) {
      await writeCsvHeader(uploadBody, columns);
    }

    uploadBody.end();
    await uploadDone;
  } catch (error) {
    uploadBody.destroy(error);
    await uploadDone.catch(() => {});
    throw new FDAError(
      503,
      'UploadError',
      `Error uploading FDA to object storage: ${error.message}`,
    );
  } finally {
    await reader.close();
  }
}

async function createMongoFDAReader(
  service,
  datasourceId,
  fdaId,
  servicePath,
  { limit } = {},
) {
  const datasource = await resolveDatasource(service, datasourceId);
  const fda = await retrieveFDA(service, fdaId, servicePath);

  if (!fda) {
    throw new FDAError(
      404,
      'FDANotFound',
      `FDA ${fdaId} not found in service ${service}`,
    );
  }

  validateMongoFDAContract(fda.query, fda.timeColumn, fda.cached);

  return await createMongoCursorReader(datasource.config, fda.query, {
    limit,
  });
}

export async function deleteFDA(service, fdaId, visibility, servicePath) {
  let targetServicePath = normalizeServicePath(servicePath);

  if (visibility !== undefined || servicePath !== undefined) {
    const fda = await getAccessibleFDA(service, fdaId, visibility, servicePath);
    targetServicePath = fda.servicePath;
  }

  const { _id } = (await retrieveFDA(service, fdaId, targetServicePath)) ?? {};

  if (!service || !_id) {
    throw new FDAError(
      404,
      'FDANotFound',
      `FDA ${fdaId} of the service ${service} not found.`,
    );
  }
  const bucketName = getBucketNameFromService(service);
  const s3Client = await getS3Client(
    `${config.objstg.protocol}://${config.objstg.endpoint}`,
    config.objstg.usr,
    config.objstg.pass,
  );
  // This way we remove FDAs independently of if theyre partitioned or not
  const storagePath = getFDAStoragePath(fdaId, targetServicePath);
  const allObjPaths = await listObjects(s3Client, bucketName, storagePath);
  // Filter strictly to objects belonging to this FDA only, preventing
  // accidental deletion of sibling FDAs whose IDs share a common prefix
  // (e.g. fda_test and fda_test_1 both match the prefix "fda_test").
  const objPaths = allObjPaths.filter(
    (key) =>
      key.startsWith(`${storagePath}/`) || key.startsWith(`${storagePath}.`),
  );
  await dropFiles(s3Client, bucketName, objPaths);

  await removeFDA(service, fdaId, targetServicePath);

  const agenda = getAgenda();
  await agenda.cancel(
    buildFDAJobCancelFilter(
      'refresh-fda-recurring',
      service,
      fdaId,
      targetServicePath,
    ),
  );
  await agenda.cancel(
    buildFDAJobCancelFilter(
      'consistency-refresh-fda-recurring',
      service,
      fdaId,
      targetServicePath,
    ),
  );
  await agenda.cancel(
    buildFDAJobCancelFilter(
      'clean-partition-recurring',
      service,
      fdaId,
      targetServicePath,
    ),
  );
}

export async function getDAs(service, fdaId, visibility, servicePath) {
  if (visibility !== undefined || servicePath !== undefined) {
    await getAccessibleFDA(service, fdaId, visibility, servicePath);
  }

  return retrieveDAs(service, fdaId, servicePath);
}

export async function getDA(service, fdaId, daId, visibility, servicePath) {
  if (visibility !== undefined || servicePath !== undefined) {
    await getAccessibleFDA(service, fdaId, visibility, servicePath);
  }

  const da = await retrieveDA(service, fdaId, daId, servicePath);
  if (da) {
    da.id = daId;
  } else {
    throw new FDAError(
      404,
      'DaNotFound',
      `DA ${daId} not found in FDA ${fdaId} and service ${service}.`,
    );
  }

  return da;
}

export async function putDA(
  service,
  fdaId,
  daId,
  description,
  userQuery,
  params,
  visibility,
  servicePath,
) {
  const conn = await getDBConnection();

  try {
    let fda;

    if (visibility !== undefined || servicePath !== undefined) {
      fda = await getAccessibleFDA(service, fdaId, visibility, servicePath);
    }

    if (!fda) {
      fda = await retrieveFDA(service, fdaId, servicePath);
    }

    const normalizedParams = checkParams(params);
    validateDAParamBindings(userQuery, normalizedParams || []);
    if (shouldValidateDACompatibility(fda)) {
      await validateDAQuery(conn, service, fdaId, userQuery, servicePath);
    }
    await updateDA(
      service,
      fdaId,
      servicePath,
      daId,
      description,
      userQuery,
      normalizedParams,
    );
  } finally {
    await releaseDBConnection(conn);
  }
}

export async function deleteDA(service, fdaId, daId, visibility, servicePath) {
  if (visibility !== undefined || servicePath !== undefined) {
    await getAccessibleFDA(service, fdaId, visibility, servicePath);
  }

  await removeDA(service, fdaId, daId, servicePath);
}

export async function cleanPartition(
  service,
  fdaId,
  windowSize,
  objStgConf,
  servicePath,
) {
  if (!objStgConf?.partition) {
    // DEBATE: With no partitioned folders doesn't make much sense to clean cause we'd had a FDA with no file
    throw new FDAError(
      400,
      'CleaningError',
      `Removing a non partitioned FDA ${fdaId}.`,
    );
  }

  const cutoff = new Date(getPreviousWindowStartDate(windowSize));
  if (!cutoff) {
    /* c8 ignore next 5 */
    throw new FDAError(
      400,
      'CleaningError',
      'Incorrect window size in refresh policy.',
    );
  }

  /* c8 ignore next 6 */
  const s3Client = getS3Client(
    `${config.objstg.protocol}://${config.objstg.endpoint}`,
    config.objstg.usr,
    config.objstg.pass,
  );
  const bucketName = getBucketNameFromService(service);

  /* c8 ignore next 10 */
  const cleanPartitionStoragePath = getFDAStoragePath(fdaId, servicePath);
  const allPartitionPaths = await listObjects(
    s3Client,
    bucketName,
    cleanPartitionStoragePath,
  );
  // Filter strictly to objects belonging to this FDA only (same prefix-collision guard as deleteFDA)
  const objPaths = allPartitionPaths.filter((key) =>
    key.startsWith(`${cleanPartitionStoragePath}/`),
  );

  const partitionsToRemove = [];
  for (const path of objPaths) {
    const partitionDate = extractDate(path);

    if (partitionDate < cutoff) {
      partitionsToRemove.push(path);
    }
  }
  await dropFiles(s3Client, bucketName, partitionsToRemove);
}

async function uploadTableToObjStg(
  service,
  datasourceId,
  query,
  bucket,
  path,
  fdaId,
  servicePath,
  timeColumn,
  objStgConf,
) {
  const s3Client = getS3Client(
    `${config.objstg.protocol}://${config.objstg.endpoint}`,
    config.objstg.usr,
    config.objstg.pass,
  );
  const datasource = await resolveDatasource(service, datasourceId);
  const fda = await retrieveFDA(service, fdaId, servicePath);
  const schemaFields = normalizePersistedSchemaFields(fda?.schema);
  await updateFDAStatus({ service, fdaId, servicePath, progress: 20 });

  if (datasource.type === 'postgres') {
    await uploadTable(s3Client, bucket, datasource.config, query, path);
  } else {
    const reader = await createMongoFDAReader(
      service,
      datasourceId,
      fdaId,
      servicePath,
    );
    await uploadMongoCursorContentToObjectStorage(
      s3Client,
      bucket,
      path,
      reader,
    );
  }

  const conn = await getDBConnection();
  try {
    await updateFDAStatus({
      service,
      fdaId,
      servicePath,
      status: 'transforming',
      progress: 60,
    });

    // DuckDB cant overwrite files in Minio, so for partitioned files we upload them in a tmp file and the move them
    // This includes first upload because the one row parquet is also partitioned
    const parquetPath = objStgConf?.partition
      ? getPath(bucket, 'tmp/' + path, '')
      : getPath(bucket, path, '.parquet');

    const csvPath = getPath(bucket, path, '.csv');
    if (datasource.type === 'postgres' && schemaFields.length > 0) {
      await copyQueryToParquet(
        conn,
        buildTypedCsvSourceQuery(csvPath, schemaFields),
        parquetPath,
        timeColumn,
        objStgConf?.partition,
        objStgConf?.compression,
      );
    } else {
      await toParquet(
        conn,
        csvPath,
        parquetPath,
        timeColumn,
        objStgConf?.partition,
        objStgConf?.compression,
      );
    }

    if (objStgConf?.partition) {
      const objectsList = await listObjects(s3Client, bucket, `tmp/${path}/`);
      const hasRealPartitionedParquet = objectsList.some((key) =>
        key.endsWith('.parquet'),
      );
      for (const tempPartition of objectsList) {
        await moveObject(
          s3Client,
          bucket,
          `${bucket}/${tempPartition}`,
          tempPartition.replace('tmp/', ''),
        );
        await dropFile(s3Client, bucket, tempPartition);
      }

      if (hasRealPartitionedParquet) {
        await dropFile(
          s3Client,
          bucket,
          `${path}/${getSchemaPartitionPath(objStgConf.partition)}/schema.parquet`,
        );
      }
    }

    await updateFDAStatus({
      service,
      fdaId,
      servicePath,
      status: 'uploading',
      progress: 80,
    });
    // DuckDb doesn't replace one row parquet snippet with partitioned file, so we remove it by hand
    if (objStgConf?.partition) {
      await dropFile(s3Client, bucket, `${path}.parquet`);
    }
    await dropFile(s3Client, bucket, `${path}.csv`);
  } catch (e) {
    throw new FDAError(500, 'UploadError', e.message);
  } finally {
    await releaseDBConnection(conn);
  }
}

async function ensureFDAReadyForQuery(service, fdaId, visibility, servicePath) {
  const fda = await getAccessibleFDA(service, fdaId, visibility, servicePath);
  assertFDAIsCached(fda, fdaId);

  // Queries are blocked only before the first successful fetch.
  if (!fda.lastFetch) {
    throw new FDAError(
      409,
      'FDAUnavailable',
      `FDA ${fdaId} is not queryable yet because the first fetch has not completed`,
    );
  }
}

export async function getStoredFDA(service, fdaId, servicePath) {
  const fda = await retrieveFDA(service, fdaId, servicePath);

  if (!fda) {
    throw new FDAError(
      404,
      'FDANotFound',
      `FDA ${fdaId} not found in service ${service}`,
    );
  }

  return fda;
}

async function getAccessibleFDA(service, fdaId, visibility, servicePath) {
  const normalizedVisibility = normalizeVisibility(visibility);
  const normalizedServicePath = normalizeServicePath(servicePath);
  const fda = await retrieveFDA(service, fdaId, normalizedServicePath);

  if (!fda) {
    const sameIdCandidates = (await retrieveFDAs(service)).filter(
      (candidate) => candidate.fdaId === fdaId,
    );

    if (sameIdCandidates.length === 0) {
      throw new FDAError(
        404,
        'FDANotFound',
        `FDA ${fdaId} not found in service ${service}`,
      );
    }

    const hasMatchingVisibility = sameIdCandidates.some(
      (candidate) =>
        normalizeVisibility(candidate.visibility) === normalizedVisibility,
    );

    if (!hasMatchingVisibility) {
      throw new FDAError(
        403,
        'VisibilityMismatch',
        `FDA ${fdaId} does not belong to ${normalizedVisibility}`,
      );
    }

    throw new FDAError(
      403,
      'ServicePathMismatch',
      `FDA ${fdaId} does not belong to servicePath ${normalizedServicePath}`,
    );
  }

  if (normalizeVisibility(fda.visibility) !== normalizedVisibility) {
    throw new FDAError(
      403,
      'VisibilityMismatch',
      `FDA ${fdaId} does not belong to ${normalizedVisibility}`,
    );
  }

  if (normalizeServicePath(fda.servicePath) !== normalizedServicePath) {
    throw new FDAError(
      403,
      'ServicePathMismatch',
      `FDA ${fdaId} does not belong to servicePath ${normalizedServicePath}`,
    );
  }

  return fda;
}

function normalizeVisibility(visibility) {
  if (!VALID_VISIBILITIES_SET.has(visibility)) {
    throw new FDAError(
      400,
      'InvalidVisibility',
      'Visibility must be public or private',
    );
  }

  return visibility;
}

function toFDAApiResponse(fda, { includeId }) {
  if (!fda) {
    return fda;
  }

  const response = { ...fda };
  const fdaId = response.fdaId;

  delete response._id;
  delete response.fdaId;
  delete response.service;
  delete response.visibility;
  delete response.servicePath;

  if (!includeId) {
    return response;
  }

  return {
    id: fdaId,
    ...response,
  };
}

async function createParquet(
  service,
  fdaId,
  query,
  servicePath,
  datasourceId,
  timeColumn,
  objStgConf,
  sourceSchema,
) {
  const s3Client = getS3Client(
    `${config.objstg.protocol}://${config.objstg.endpoint}`,
    config.objstg.usr,
    config.objstg.pass,
  );
  const storagePath = getFDAStoragePath(fdaId, servicePath);
  const bucketName = getBucketNameFromService(service);
  const datasource = await resolveDatasource(service, datasourceId);
  const parquetPath = getPath(bucketName, storagePath, '.parquet');

  if (datasource.type === 'postgres') {
    const typedEmptyQuery = buildTypedEmptyQueryFromSchema(sourceSchema);

    if (typedEmptyQuery) {
      const conn = await getDBConnection();
      try {
        await copyQueryToParquet(
          conn,
          typedEmptyQuery,
          parquetPath,
          timeColumn,
          objStgConf?.partition,
          objStgConf?.compression,
        );
      } finally {
        await releaseDBConnection(conn);
      }

      return;
    }

    const oneRowQuery = buildZeroRowQuery(query);
    await uploadTable(
      s3Client,
      bucketName,
      datasource.config,
      oneRowQuery,
      storagePath,
    );
  } else {
    const reader = await createMongoFDAReader(
      service,
      datasourceId,
      fdaId,
      servicePath,
      { limit: 1 },
    );
    await uploadMongoCursorContentToObjectStorage(
      s3Client,
      bucketName,
      storagePath,
      reader,
    );
  }

  const conn = await getDBConnection();
  try {
    await toParquet(
      conn,
      getPath(bucketName, storagePath, '.csv'),
      parquetPath,
      timeColumn,
      objStgConf?.partition,
      objStgConf?.compression,
    );

    await dropFile(s3Client, bucketName, `${storagePath}.csv`);
  } finally {
    await releaseDBConnection(conn);
  }
}

function buildZeroRowQuery(query) {
  const normalizedQuery = query.trim().replace(/;+\s*$/, '');

  // Schema-only bootstrap keeps creation validation synchronous without row materialization.
  return `SELECT * FROM (${normalizedQuery}) AS fda_one_row LIMIT 0`;
}

function isValidDuckDBType(type) {
  return /^[A-Za-z0-9_,()\s]+$/.test(type);
}

function buildTypedEmptyQueryFromSchema(sourceSchema) {
  const fields = Array.isArray(sourceSchema?.fields)
    ? sourceSchema.fields.filter(
        (field) =>
          typeof field?.name === 'string' &&
          field.name.length > 0 &&
          typeof field?.duckdbType === 'string' &&
          field.duckdbType.length > 0,
      )
    : [];

  if (fields.length === 0) {
    return null;
  }

  const projections = fields.map(({ name, duckdbType }) => {
    if (!isValidDuckDBType(duckdbType)) {
      throw new FDAError(
        500,
        'InvalidSchemaType',
        `Invalid schema type for column ${name}: ${duckdbType}`,
      );
    }

    return `CAST(NULL AS ${duckdbType}) AS ${quoteDuckDBIdentifier(name)}`;
  });

  return `SELECT ${projections.join(', ')} WHERE FALSE`;
}

function normalizePersistedSchemaFields(schema) {
  if (!Array.isArray(schema)) {
    return [];
  }

  return schema.filter(
    (field) =>
      typeof field?.name === 'string' &&
      field.name.length > 0 &&
      typeof field?.type === 'string' &&
      field.type.length > 0 &&
      isValidDuckDBType(field.type),
  );
}

function quoteSqlStringLiteral(value) {
  return `'${String(value).replaceAll("'", "''")}'`;
}

function buildTypedCsvSourceQuery(csvPath, schemaFields) {
  const columnsConf = schemaFields
    .map(
      ({ name, type }) =>
        `${quoteSqlStringLiteral(name)}: ${quoteSqlStringLiteral(type)}`,
    )
    .join(', ');

  return `SELECT * FROM read_csv('s3://${csvPath}', header = true, columns = {${columnsConf}})`;
}

async function buildDefaultDataAccessDefinition(
  service,
  fdaId,
  servicePath,
  timeColumn,
  objStgConf,
  schemaOverride,
) {
  let overrideColumns = [];

  if (Array.isArray(schemaOverride?.columns)) {
    overrideColumns = schemaOverride.columns;
  } else if (Array.isArray(schemaOverride)) {
    overrideColumns = schemaOverride
      .map((column) => (typeof column === 'string' ? column : column?.name))
      .filter((column) => typeof column === 'string');
  }

  const normalizedOverrideColumns = overrideColumns.filter(
    (name) => typeof name === 'string' && name.length > 0,
  );

  const columns =
    normalizedOverrideColumns.length > 0
      ? normalizedOverrideColumns
      : await getFDAColumnNamesFromStorage(
          service,
          fdaId,
          servicePath,
          objStgConf,
        );

  const resolvedTimeColumn = resolveDefaultDATimeColumnName(
    timeColumn,
    columns,
  );
  const reservedParamNames = ['pageSize', 'pageStart'];
  if (resolvedTimeColumn) {
    reservedParamNames.push('start', 'finish');
  }

  if (columns.length === 0) {
    const params = reservedParamNames.map((name) => {
      if (name === 'pageSize') {
        return { name, default: '9223372036854775807' };
      }

      if (name === 'pageStart') {
        return { name, default: 0 };
      }

      return { name, default: null };
    });

    return {
      query:
        'SELECT *, COUNT(*) OVER() as __total LIMIT CAST($pageSize AS BIGINT) OFFSET CAST($pageStart AS BIGINT)',
      params,
    };
  }

  const usedParamNames = new Set(reservedParamNames);
  const params = [];
  const filters = [];

  for (const columnName of columns) {
    const baseName = sanitizeDefaultDAParamBaseName(columnName);
    const paramName = getUniqueDefaultDAParamName(baseName, usedParamNames);
    const quotedColumnName = quoteDuckDBIdentifier(columnName);
    const isResolvedTimeColumn =
      resolvedTimeColumn && columnName === resolvedTimeColumn;

    params.push({ name: paramName, default: null });
    if (isResolvedTimeColumn) {
      filters.push(
        `($${paramName} IS NULL OR DATE_TRUNC('millisecond', CAST(${quotedColumnName} AS TIMESTAMP)) = DATE_TRUNC('millisecond', CAST($${paramName} AS TIMESTAMP)))`,
      );
    } else {
      filters.push(
        `($${paramName} IS NULL OR ${quotedColumnName} = $${paramName})`,
      );
    }
  }

  if (resolvedTimeColumn) {
    const quotedTimeColumn = quoteDuckDBIdentifier(resolvedTimeColumn);
    params.push({ name: 'start', default: null });
    params.push({ name: 'finish', default: null });
    filters.push(
      `($start IS NULL OR CAST(${quotedTimeColumn} AS TIMESTAMP) >= CAST($start AS TIMESTAMP))`,
    );
    filters.push(
      `($finish IS NULL OR CAST(${quotedTimeColumn} AS TIMESTAMP) <= CAST($finish AS TIMESTAMP))`,
    );
  }

  params.push({ name: 'pageSize', default: '9223372036854775807' });
  params.push({ name: 'pageStart', default: 0 });

  const whereClause =
    filters.length > 0 ? ` WHERE ${filters.join(' AND ')}` : '';

  return {
    query: `SELECT *, COUNT(*) OVER() as __total${whereClause} LIMIT CAST($pageSize AS BIGINT) OFFSET CAST($pageStart AS BIGINT)`,
    params,
  };
}

function resolveDefaultDATimeColumnName(timeColumn, columns) {
  if (typeof timeColumn !== 'string' || timeColumn.length === 0) {
    return null;
  }

  const exactMatch = columns.find((column) => column === timeColumn);
  if (exactMatch) {
    return exactMatch;
  }

  const lowerTimeColumn = timeColumn.toLowerCase();
  return (
    columns.find(
      (column) =>
        typeof column === 'string' && column.toLowerCase() === lowerTimeColumn,
    ) || null
  );
}

async function getFDAColumnNamesFromStorage(
  service,
  fdaId,
  servicePath,
  objStgConf,
) {
  const conn = await getDBConnection();
  try {
    const storagePath = getFDAStoragePath(fdaId, servicePath);
    const bucketName = getBucketNameFromService(service);
    const parquetPath = objStgConf?.partition
      ? `s3://${bucketName}/${storagePath}.parquet/**/*.parquet`
      : `s3://${bucketName}/${storagePath}.parquet`;
    const safeParquetPath = parquetPath.replaceAll("'", "''");
    const describeResult = await conn.run(
      `DESCRIBE SELECT * FROM read_parquet('${safeParquetPath}')`,
    );

    const describeRows = await Promise.resolve(
      describeResult.getRowObjectsJson(),
    );

    return describeRows
      .map(
        (row) =>
          row?.column_name ?? row?.columnName ?? row?.name ?? row?.column,
      )
      .filter((name) => typeof name === 'string' && name.length > 0);
  } finally {
    await releaseDBConnection(conn);
  }
}

function getSchemaPartitionPath(partitionType) {
  const partitionPaths = {
    day: 'year=9999/month=12/day=31',
    week: 'year=9999/week=9999-52',
    month: 'year=9999/month=12',
    year: 'year=9999',
  };

  return partitionPaths[partitionType] ?? partitionPaths.year;
}

function quoteDuckDBIdentifier(identifier) {
  return `"${String(identifier).replace(/"/g, '""')}"`;
}

function sanitizeDefaultDAParamBaseName(columnName) {
  const normalized = String(columnName)
    .toLowerCase()
    .replace(/[^a-z0-9_]/g, '_')
    .replace(/_+/g, '_')
    .replace(/^_+|_+$/g, '');

  if (!normalized) {
    return 'col';
  }

  if (/^[0-9]/.test(normalized)) {
    return `col_${normalized}`;
  }

  return normalized;
}

function getUniqueDefaultDAParamName(baseName, usedParamNames) {
  let candidate = baseName;
  let suffix = 2;

  while (usedParamNames.has(candidate)) {
    candidate = `${baseName}_${suffix}`;
    suffix += 1;
  }

  usedParamNames.add(candidate);
  return candidate;
}

async function rollbackFDAProvisioning(service, fdaId, servicePath) {
  const s3Client = getS3Client(
    `${config.objstg.protocol}://${config.objstg.endpoint}`,
    config.objstg.usr,
    config.objstg.pass,
  );
  const storagePath = getFDAStoragePath(fdaId, servicePath);
  const bucketName = getBucketNameFromService(service);

  const rollbackResults = await Promise.allSettled([
    dropFile(s3Client, bucketName, `${storagePath}.csv`),
    dropFile(s3Client, bucketName, `${storagePath}.parquet`),
    removeFDA(service, fdaId, servicePath),
  ]);

  const mongoRollbackResult = rollbackResults[2];
  if (mongoRollbackResult.status === 'rejected') {
    throw mongoRollbackResult.reason;
  }
}

function assertFDAIsCached(fda, fdaId) {
  if (fda?.cached === false) {
    throw new FDAError(
      409,
      'FDAOnlyFresh',
      `FDA ${fdaId} is configured as only-fresh and does not support this operation.`,
    );
  }
}

const getPath = (bucket, path, extension) => {
  const cleanBucket = bucket?.endsWith('/') ? bucket.slice(0, -1) : bucket;
  const cleanPath = path?.startsWith('/') ? path.slice(1) : path;
  return `${cleanBucket}/${cleanPath}${extension}`;
};

async function scheduleFDAJobs({
  fdaId,
  query, // Query original (sin filtrar)
  service,
  servicePath,
  timeColumn,
  refreshPolicy,
  objStgConf,
  datasource,
}) {
  const { refreshInterval, windowSize, consistencyRefreshInterval } =
    refreshPolicy.params || {};

  const agenda = getAgenda();

  const { firstQuery, recurringQuery } = resolveRefreshQueries(
    datasource,
    refreshPolicy,
    query,
    timeColumn,
  );

  // 0. Execute first fetch immediately (fetch all data inside window if windowSize is defined, otherwise fetch all data)
  await agenda.now('refresh-fda', {
    fdaId,
    query: firstQuery,
    service,
    servicePath,
    timeColumn,
    refreshPolicy,
    objStgConf,
    datasourceId: datasource.datasourceId ?? DEFAULT_DATASOURCE_ID,
  });

  if (refreshPolicy?.type === 'interval' || refreshPolicy?.type === 'window') {
    // 1. Recurring fetch job (incremental fetches)
    const refreshJob = agenda.create('refresh-fda-recurring', {
      fdaId,
      query: recurringQuery,
      service,
      servicePath,
      timeColumn,
      refreshPolicy,
      objStgConf,
      datasourceId: datasource.datasourceId ?? DEFAULT_DATASOURCE_ID,
    });

    refreshJob.unique(
      buildFDAJobFilter('refresh-fda-recurring', service, fdaId, servicePath),
    );
    refreshJob.repeatEvery(refreshInterval, { skipImmediate: true });
    await refreshJob.save();

    // 2. Consistency refresh job if applicable (fech all data inside windowSize if windowSize is defined, otherwise fetch all data)
    if (consistencyRefreshInterval) {
      const consistencyRefreshJob = agenda.create(
        'consistency-refresh-fda-recurring',
        {
          fdaId,
          query: firstQuery,
          service,
          servicePath,
          timeColumn,
          refreshPolicy,
          objStgConf,
          datasourceId: datasource.datasourceId ?? DEFAULT_DATASOURCE_ID,
        },
      );

      consistencyRefreshJob.unique(
        buildFDAJobFilter(
          'consistency-refresh-fda-recurring',
          service,
          fdaId,
          servicePath,
        ),
      );
      consistencyRefreshJob.repeatEvery(consistencyRefreshInterval, {
        skipImmediate: true,
      });
      await consistencyRefreshJob.save();
    }

    // 3. Clean partition job if applicable (remove all data older than windowSize)
    if (windowSize) {
      const cleanPartitionJob = agenda.create('clean-partition-recurring', {
        fdaId,
        service,
        servicePath,
        windowSize,
        objStgConf,
      });

      cleanPartitionJob.unique(
        buildFDAJobFilter(
          'clean-partition-recurring',
          service,
          fdaId,
          servicePath,
        ),
      );
      cleanPartitionJob.repeatEvery(refreshInterval, { skipImmediate: true });
      await cleanPartitionJob.save();
    }
  }
}

async function createDefaultDAIfNeeded({
  cached,
  defaultDataAccessEnabled,
  service,
  fdaId,
  normalizedServicePath,
  timeColumn,
  objStgConf,
  normalizedVisibility,
  sourceSchema,
}) {
  if (!cached || !defaultDataAccessEnabled) {
    return;
  }

  try {
    const defaultDA = await buildDefaultDataAccessDefinition(
      service,
      fdaId,
      normalizedServicePath,
      timeColumn,
      objStgConf,
      sourceSchema,
    );

    await createDA(
      service,
      fdaId,
      'defaultDataAccess',
      'Default Data Access providing access to whole FDA data. It has parameters for all columns in the FDA.',
      defaultDA.query,
      defaultDA.params,
      normalizedVisibility,
      normalizedServicePath,
    );
  } catch (err) {
    await rollbackFDAProvisioning(service, fdaId, normalizedServicePath);
    throw new FDAError(
      500,
      'DefaultDataAccessCreationError',
      `Failed to create default Data Access for FDA ${fdaId}: ${err.message}`,
    );
  }
}

telefonicaid / fiware-data-access / 30084067687

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous