package jnpf.limsService.provider;
|
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
import com.fasterxml.jackson.databind.node.ObjectNode;
|
import jnpf.exception.DataException;
|
import jnpf.lims.api.model.TestWorkItemSnapshot;
|
import jnpf.lims.api.model.TestWorkItemSnapshotRequest;
|
import jnpf.limsService.LimsTestWorkItemSnapshotService;
|
import jnpf.limsService.provider.LimsElnReadonlyResolver.Field;
|
import jnpf.limsService.provider.LimsElnReadonlyResolver.FrozenDefinition;
|
import jnpf.limsService.provider.LimsElnReadonlyResolver.Join;
|
import jnpf.limsService.provider.LimsElnReadonlyResolver.Snapshot;
|
import jnpf.limsService.provider.LimsElnReadonlyResolver.SourceFailure;
|
import jnpf.limsService.provider.LimsElnReadonlyResolver.Condition;
|
|
import javax.sql.DataSource;
|
import java.sql.Connection;
|
import java.sql.PreparedStatement;
|
import java.sql.ResultSet;
|
import java.sql.SQLException;
|
import java.sql.Types;
|
import java.time.Instant;
|
import java.util.ArrayList;
|
import java.util.Collections;
|
import java.util.LinkedHashMap;
|
import java.util.LinkedHashSet;
|
import java.util.List;
|
import java.util.Map;
|
import java.util.Set;
|
import java.util.stream.Collectors;
|
|
/** Package-local query implementation; physical mappings are never HTTP inputs. */
|
final class LimsElnSourceExecution {
|
private final DataSource readOnlyDataSource;
|
private final LimsTestWorkItemSnapshotService taskAccess;
|
private final ObjectMapper json = new ObjectMapper();
|
|
LimsElnSourceExecution(DataSource readOnlyDataSource, LimsTestWorkItemSnapshotService taskAccess) {
|
this.readOnlyDataSource = readOnlyDataSource;
|
this.taskAccess = taskAccess;
|
}
|
|
List<Snapshot> execute(List<String> itemIds, List<LimsElnSourceCatalog.Version> versions,
|
LimsElnSourceCatalog.Actor actor) {
|
if (actor == null || !"0".equals(actor.getTenantId()) || actor.getUserId() == null
|
|| actor.getUserId().trim().isEmpty()) throw new DataException("Invalid source caller");
|
if (itemIds == null || versions == null || itemIds.size() > 100 || versions.size() > 50) {
|
throw new DataException("Source request exceeds limits");
|
}
|
if (itemIds.stream().anyMatch(id -> id == null || id.trim().isEmpty() || id.length() > 128)) {
|
throw new DataException("Invalid source request");
|
}
|
itemIds = new ArrayList<>(new LinkedHashSet<>(itemIds));
|
versions = new ArrayList<>(new LinkedHashSet<>(versions));
|
if (itemIds.isEmpty() || versions.isEmpty()) return Collections.emptyList();
|
TestWorkItemSnapshotRequest accessRequest = new TestWorkItemSnapshotRequest();
|
accessRequest.setTestWorkItemIds(itemIds);
|
accessRequest.setProperties(Collections.emptyList());
|
Map<String, String> access = new LinkedHashMap<>();
|
for (TestWorkItemSnapshot item : taskAccess.resolve(accessRequest, "0", actor.getUserId()).getItems()) {
|
access.put(item.getTestWorkItemId(), item.getOutcomeCode());
|
}
|
List<String> authorized = itemIds.stream().filter(id -> "SUCCESS".equals(access.get(id)))
|
.collect(Collectors.toList());
|
Map<String, List<LimsElnSourceCatalog.Version>> groups = new LinkedHashMap<>();
|
Map<LimsElnSourceCatalog.Version, FrozenDefinition> definitions = new LinkedHashMap<>();
|
for (LimsElnSourceCatalog.Version version : versions) {
|
try {
|
FrozenDefinition definition = json.readValue(version.getFrozenDefinitionJson(), FrozenDefinition.class);
|
if (definition.getDatasetCode() == null || definition.getDatasetRevision() < 1) {
|
throw new DataException("Invalid source dataset version");
|
}
|
definitions.put(version, definition);
|
groups.computeIfAbsent(definition.getDatasetCode() + "@" + definition.getDatasetRevision(),
|
key -> new ArrayList<>()).add(version);
|
} catch (java.io.IOException e) {
|
throw new DataException("Source definition is invalid");
|
}
|
}
|
if (groups.size() > 10) throw new DataException("Source request exceeds limits");
|
Map<LimsElnSourceCatalog.Version, Map<String, Snapshot>> resolved = new LinkedHashMap<>();
|
for (List<LimsElnSourceCatalog.Version> group : groups.values()) {
|
resolved.putAll(read(group, definitions, authorized));
|
}
|
List<Snapshot> result = new ArrayList<>();
|
for (LimsElnSourceCatalog.Version version : definitions.keySet()) {
|
for (String id : itemIds) {
|
if (!"SUCCESS".equals(access.get(id))) {
|
result.add(new Snapshot(id, version.getPropertyCode(), version.getRevision(),
|
version.getDataType(), "NOT_AVAILABLE", null, null, null));
|
continue;
|
}
|
result.add(resolved.get(version).get(id));
|
}
|
}
|
return result;
|
}
|
|
private Map<LimsElnSourceCatalog.Version, Map<String, Snapshot>> read(
|
List<LimsElnSourceCatalog.Version> versions,
|
Map<LimsElnSourceCatalog.Version, FrozenDefinition> definitions, List<String> itemIds) {
|
Map<LimsElnSourceCatalog.Version, Map<String, Snapshot>> result = new LinkedHashMap<>();
|
if (itemIds.isEmpty()) return result;
|
try {
|
FrozenDefinition definition = definitions.get(versions.get(0));
|
if (definition.getJoins() == null || definition.getJoins().size() > 4
|
|| definition.getFields() == null || definition.getFields().size() > 50) {
|
throw new DataException("Source request exceeds limits");
|
}
|
ObjectNode dataset = json.valueToTree(definition);
|
dataset.remove("resultField");
|
List<Field> fields = new ArrayList<>();
|
for (LimsElnSourceCatalog.Version version : versions) {
|
FrozenDefinition candidate = definitions.get(version);
|
ObjectNode candidateDataset = json.valueToTree(candidate);
|
candidateDataset.remove("resultField");
|
if (!dataset.equals(candidateDataset)) throw new DataException("Conflicting source dataset version");
|
Field field = candidate.getFields().get(candidate.getResultField());
|
if (field == null || !version.getDataType().equals(field.getDataType())) {
|
throw new DataException("Invalid source result field");
|
}
|
if (field.getStep() < 0 || field.getStep() > definition.getJoins().size()) {
|
throw new DataException("Invalid source relation reference");
|
}
|
fields.add(field);
|
}
|
String table = identifier(definition.getSchema()) + "." + identifier(definition.getTable());
|
StringBuilder from = new StringBuilder(table).append(" t0");
|
List<String> joinParameters = new ArrayList<>();
|
for (int i = 0; i < definition.getJoins().size(); i++) {
|
Join join = definition.getJoins().get(i);
|
if (join.getFromStep() < 0 || join.getFromStep() > i) {
|
throw new DataException("Invalid source relation reference");
|
}
|
String target = identifier(join.getSchema()) + "." + identifier(join.getTable());
|
from.append(" LEFT JOIN ").append(target).append(" t").append(i + 1)
|
.append(" ON t").append(i + 1).append('.').append(identifier(join.getTenantColumn()))
|
.append(" = t0.").append(identifier(definition.getTenantColumn()))
|
.append(" AND t").append(i + 1).append('.').append(identifier(join.getTargetColumn()))
|
.append(" = t").append(join.getFromStep()).append('.').append(identifier(join.getFromColumn()));
|
if (join.getDeleteColumn() != null) {
|
from.append(" AND ").append(notDeleted("t" + (i + 1), join.getDeleteColumn()));
|
}
|
if (join.getQualifierColumn() != null) {
|
if (join.getQualifierValue() == null || join.getQualifierValue().length() > 200) {
|
throw new DataException("Invalid source relation qualifier");
|
}
|
from.append(" AND t").append(i + 1).append('.').append(identifier(join.getQualifierColumn())).append(" = ?");
|
joinParameters.add(join.getQualifierValue());
|
}
|
}
|
String selected = fields.stream().map(LimsElnSourceExecution::selectedExpression)
|
.collect(Collectors.joining(", "));
|
List<String> scopeParameters = new ArrayList<>();
|
String scope = scopePredicate(definition, scopeParameters);
|
String sql = "SELECT t0." + identifier(definition.getIdColumn()) + ", " + selected
|
+ ", (" + scope + ") FROM " + from
|
+ " WHERE " + (definition.getDeleteColumn() == null ? "" : notDeleted("t0", definition.getDeleteColumn()) + " AND ")
|
+ "t0." + identifier(definition.getTenantColumn()) + " = ? AND t0."
|
+ identifier(definition.getIdColumn()) + " IN ("
|
+ String.join(",", Collections.nCopies(itemIds.size(), "?")) + ") LIMIT " + (itemIds.size() * 2 + 1);
|
try (Connection connection = readOnlyDataSource.getConnection()) {
|
connection.setReadOnly(true);
|
connection.setTransactionIsolation(Connection.TRANSACTION_REPEATABLE_READ);
|
connection.setAutoCommit(false);
|
try (PreparedStatement query = connection.prepareStatement(sql)) {
|
enforceReadOnlyTransaction(connection);
|
query.setQueryTimeout(5);
|
query.setMaxRows(itemIds.size() * 2 + 1);
|
int parameter = 1;
|
for (String scopeValue : scopeParameters) query.setString(parameter++, scopeValue);
|
for (String joinValue : joinParameters) query.setString(parameter++, joinValue);
|
query.setString(parameter++, "0");
|
for (String id : itemIds) query.setString(parameter++, id);
|
try (ResultSet rows = query.executeQuery()) {
|
for (int i = 0; i < fields.size(); i++) {
|
if (!compatible(fields.get(i).getDataType(), rows.getMetaData().getColumnType(i + 2))) {
|
throw new DataException("Source physical type has changed");
|
}
|
}
|
Map<String, List<Object>> values = new LinkedHashMap<>();
|
// The SELECT holds relation locks until this read transaction ends, preventing DDL races.
|
validateColumnIdentities(connection, definition, versions.stream()
|
.map(version -> definitions.get(version).getResultField()).collect(Collectors.toSet()));
|
Map<String, Boolean> applicable = new LinkedHashMap<>();
|
while (rows.next()) {
|
String id = rows.getString(1);
|
if (!itemIds.contains(id)) continue;
|
if (values.containsKey(id)) {
|
throw new SourceFailure("SOURCE_AMBIGUOUS", "Source is not single-valued");
|
}
|
List<Object> columns = new ArrayList<>();
|
for (int i = 0; i < fields.size(); i++) {
|
Object value = "timestamptz".equals(rows.getMetaData().getColumnTypeName(i + 2))
|
? rows.getObject(i + 2, java.time.OffsetDateTime.class) : rows.getObject(i + 2);
|
if (value instanceof String && ((String) value).trim().isEmpty()) value = null;
|
columns.add(value);
|
}
|
values.put(id, columns);
|
applicable.put(id, rows.getBoolean(fields.size() + 2));
|
}
|
String acquiredAt = Instant.now().toString();
|
for (int i = 0; i < versions.size(); i++) {
|
LimsElnSourceCatalog.Version version = versions.get(i);
|
Map<String, Snapshot> snapshots = new LinkedHashMap<>();
|
for (String id : itemIds) {
|
Object value = values.containsKey(id) ? values.get(id).get(i) : null;
|
boolean inScope = applicable.getOrDefault(id, true);
|
Field field = fields.get(i);
|
String reason = null;
|
if (inScope && "DISPLAY_NAME".equals(field.getValueMode())) {
|
if (field.getNameOptions() != null) value = field.getNameOptions().get(value);
|
if (value == null) reason = "NAME_MISSING";
|
}
|
snapshots.put(id, new Snapshot(id, version.getPropertyCode(), version.getRevision(),
|
version.getDataType(), !inScope ? "NOT_APPLICABLE" : value == null ? "MISSING" : "SUCCESS",
|
inScope ? value : null,
|
acquiredAt, definition.getDatasetCode() + "@" + definition.getDatasetRevision()
|
+ "/" + definitions.get(version).getResultField(), reason,
|
inScope ? formattedValue(field, value) : null));
|
}
|
result.put(version, snapshots);
|
}
|
return result;
|
}
|
} finally {
|
connection.rollback();
|
}
|
}
|
} catch (SourceFailure e) {
|
throw e;
|
} catch (DataException e) {
|
throw new SourceFailure("SOURCE_INVALID", e.getMessage());
|
} catch (SQLException e) {
|
String state = e.getSQLState();
|
if (state != null && (state.startsWith("42") || state.startsWith("22"))) {
|
throw new SourceFailure("42501".equals(state) ? "SOURCE_FORBIDDEN" : "SOURCE_INVALID",
|
"Source configuration is no longer readable");
|
}
|
if ("57014".equals(state)) throw new SourceFailure("SOURCE_LIMIT", "Source query timed out");
|
throw new SourceFailure("SOURCE_UNAVAILABLE", "Source database is unavailable");
|
} catch (Exception e) {
|
throw new SourceFailure("SOURCE_INVALID", "Source definition or read failed");
|
}
|
}
|
|
private static String formattedValue(Field field, Object value) {
|
if (value == null || "RAW".equals(field.getFormat())) return null;
|
if ("PLAIN_DECIMAL".equals(field.getFormat()) && "DECIMAL".equals(field.getDataType()) && value instanceof Number) {
|
return value instanceof java.math.BigDecimal ? ((java.math.BigDecimal) value).toPlainString() : value.toString();
|
}
|
if (("ISO_DATE".equals(field.getFormat()) || "DATE_ONLY".equals(field.getFormat()))
|
&& "DATE".equals(field.getDataType()) && value instanceof java.sql.Date) {
|
return ((java.sql.Date) value).toLocalDate().toString();
|
}
|
throw new DataException("Invalid source format for field type");
|
}
|
|
private static String selectedExpression(Field field) {
|
String column = "t" + field.getStep() + "." + identifier(field.getColumn());
|
return "DATE_ONLY".equals(field.getFormat()) ? "CAST(" + column + " AS date)" : column;
|
}
|
|
static String identifier(String value) {
|
if (value == null || !value.matches("[a-z_][a-z0-9_]{0,62}")) {
|
throw new DataException("Invalid source identifier");
|
}
|
return "\"" + value + "\"";
|
}
|
|
private static void validateColumnIdentities(Connection connection, FrozenDefinition definition,
|
Set<String> fields) throws SQLException {
|
Map<String, Map<String, String>> expected = definition.getColumnIdentities();
|
Map<String, Map<String, String>> actual = columnIdentities(connection, definition, fields);
|
if (expected == null || actual.entrySet().stream().anyMatch(table -> expected.get(table.getKey()) == null
|
|| !expected.get(table.getKey()).entrySet().containsAll(table.getValue().entrySet()))) {
|
throw new SourceFailure("SOURCE_INVALID", "Source physical identity has changed");
|
}
|
}
|
|
static Map<String, Map<String, String>> columnIdentities(Connection connection, FrozenDefinition definition,
|
Set<String> fieldKeys) throws SQLException {
|
List<String> tables = new ArrayList<>();
|
tables.add(identifier(definition.getSchema()) + "." + identifier(definition.getTable()));
|
Map<String, Set<String>> references = new LinkedHashMap<>();
|
addReference(references, tables.get(0), definition.getIdColumn());
|
addReference(references, tables.get(0), definition.getTenantColumn());
|
addReference(references, tables.get(0), definition.getDeleteColumn());
|
for (Join join : definition.getJoins()) {
|
String target = identifier(join.getSchema()) + "." + identifier(join.getTable());
|
addReference(references, tables.get(join.getFromStep()), join.getFromColumn());
|
addReference(references, target, join.getTargetColumn());
|
addReference(references, target, join.getTenantColumn());
|
addReference(references, target, join.getDeleteColumn());
|
addReference(references, target, join.getQualifierColumn());
|
tables.add(target);
|
}
|
Set<String> required = new LinkedHashSet<>(fieldKeys);
|
for (Condition condition : definition.getConditions()) required.add(condition.getField());
|
for (String key : required) {
|
Field field = definition.getFields().get(key);
|
addReference(references, tables.get(field.getStep()), field.getColumn());
|
}
|
Map<String, Map<String, String>> result = new LinkedHashMap<>();
|
for (Map.Entry<String, Set<String>> table : references.entrySet()) {
|
Map<String, String> columns = new LinkedHashMap<>();
|
try (PreparedStatement query = connection.prepareStatement("SELECT attname, "
|
+ "'c' || attrelid::text || '_' || attnum::text || '_' || atttypid::text FROM pg_catalog.pg_attribute "
|
+ "WHERE attrelid = ?::regclass AND attnum > 0 AND NOT attisdropped")) {
|
query.setQueryTimeout(5);
|
query.setString(1, table.getKey());
|
try (ResultSet rows = query.executeQuery()) {
|
while (rows.next()) {
|
if (table.getValue().contains(rows.getString(1))) columns.put(rows.getString(1), rows.getString(2));
|
}
|
}
|
}
|
if (!columns.keySet().equals(table.getValue())) throw new DataException("Source dependency is no longer available");
|
result.put(table.getKey(), columns);
|
}
|
return result;
|
}
|
|
private static void addReference(Map<String, Set<String>> references, String table, String column) {
|
if (column == null) return;
|
identifier(column);
|
references.computeIfAbsent(table, ignored -> new LinkedHashSet<>()).add(column);
|
}
|
|
private static String notDeleted(String alias, String column) {
|
String field = alias + "." + identifier(column);
|
return "(" + field + " IS NULL OR " + field + " = 0)";
|
}
|
|
static void validateScope(FrozenDefinition definition) {
|
scopePredicate(definition, new ArrayList<>());
|
}
|
|
private static String scopePredicate(FrozenDefinition definition, List<String> parameters) {
|
if (definition.getConditions() == null || definition.getConditions().size() > 20
|
|| !("ALL".equals(definition.getScopeMode()) || "ANY".equals(definition.getScopeMode()))) {
|
throw new DataException("Invalid source scope");
|
}
|
List<String> predicates = new ArrayList<>();
|
for (Condition condition : definition.getConditions()) {
|
if (condition == null) throw new DataException("Invalid source scope field");
|
Field field = definition.getFields().get(condition.getField());
|
if (field == null || field.getStep() < 0 || field.getStep() > definition.getJoins().size()) {
|
throw new DataException("Invalid source scope field");
|
}
|
if (!"RAW".equals(field.getValueMode())) throw new DataException("Source scope requires raw fields");
|
String column = "t" + field.getStep() + "." + identifier(field.getColumn());
|
if ("IS_NULL".equals(condition.getOperator()) || "IS_NOT_NULL".equals(condition.getOperator())) {
|
predicates.add(column + ("IS_NULL".equals(condition.getOperator()) ? " IS NULL" : " IS NOT NULL"));
|
continue;
|
}
|
if (!("EQ".equals(condition.getOperator()) || "IN".equals(condition.getOperator()))
|
|| !"TEXT".equals(field.getDataType()) || condition.getValues() == null
|
|| condition.getValues().isEmpty() || condition.getValues().size() > 100
|
|| "EQ".equals(condition.getOperator()) && condition.getValues().size() != 1
|
|| condition.getValues().stream().anyMatch(value -> value == null || value.length() > 200)) {
|
throw new DataException("Invalid source scope comparison");
|
}
|
predicates.add(column + " IN (" + String.join(",", Collections.nCopies(condition.getValues().size(), "?")) + ")");
|
parameters.addAll(condition.getValues());
|
}
|
return predicates.isEmpty() ? "TRUE" : String.join("ALL".equals(definition.getScopeMode()) ? " AND " : " OR ", predicates);
|
}
|
|
static void enforceReadOnlyTransaction(Connection connection) throws SQLException {
|
// JDBC readOnlyMode=ignore must not disable the execution transaction's write protection.
|
try (java.sql.Statement statement = connection.createStatement()) {
|
statement.setQueryTimeout(5);
|
statement.execute("SET TRANSACTION READ ONLY");
|
}
|
}
|
|
private static boolean compatible(String type, int jdbcType) {
|
switch (type) {
|
case "TEXT": return jdbcType == Types.VARCHAR || jdbcType == Types.CHAR || jdbcType == Types.LONGVARCHAR;
|
case "DECIMAL": return jdbcType == Types.NUMERIC || jdbcType == Types.DECIMAL
|
|| jdbcType == Types.INTEGER || jdbcType == Types.BIGINT || jdbcType == Types.SMALLINT;
|
case "DATE": return jdbcType == Types.DATE;
|
case "DATETIME": return jdbcType == Types.TIMESTAMP || jdbcType == Types.TIMESTAMP_WITH_TIMEZONE;
|
case "BOOLEAN": return jdbcType == Types.BOOLEAN || jdbcType == Types.BIT;
|
default: return false;
|
}
|
}
|
|
}
|
