package com.cloudhandson.vpdbackoffice.service;

import com.cloudhandson.vpdbackoffice.domain.audit.AuditEvent;
import com.cloudhandson.vpdbackoffice.domain.probe.ProbeCommand;
import com.cloudhandson.vpdbackoffice.domain.probe.ProbeResult;
import com.cloudhandson.vpdbackoffice.domain.probe.ProbeStatus;
import com.cloudhandson.vpdbackoffice.domain.probe.FgaExecutionEvidence;
import com.cloudhandson.vpdbackoffice.domain.probe.SqlExecutionCandidate;
import com.cloudhandson.vpdbackoffice.domain.probe.SqlExecutionEvidence;
import com.cloudhandson.vpdbackoffice.domain.protectedobject.ProtectedColumn;
import com.cloudhandson.vpdbackoffice.domain.protectedobject.ProtectedObject;
import com.cloudhandson.vpdbackoffice.domain.token.BearerTokenRecord;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.net.URI;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.time.Clock;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jdbc.core.ConnectionCallback;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.web.client.HttpStatusCodeException;
import org.springframework.web.client.ResourceAccessException;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.util.UriComponentsBuilder;

@Service
public class OrdsProbeService {

  private static final TypeReference<Map<String, Object>> MAP_TYPE = new TypeReference<>() {
  };
  private static final Logger log = LoggerFactory.getLogger(OrdsProbeService.class);

  private record ExecutionEvidenceLookup(
      SqlExecutionEvidence evidence,
      List<SqlExecutionCandidate> candidates,
      boolean sqlCatalogAccessUnavailable
  ) {
  }

  private record RecentSqlCursor(
      String sqlId,
      int childNumber,
      String parsingSchema,
      String sqlFulltext,
      String lastActiveAt,
      long executions,
      long rowsProcessed,
      long elapsedMillis,
      long bufferGets
  ) {
  }

  private final BearerTokenService tokenService;
  private final ProtectedObjectService protectedObjectService;
  private final AuditService auditService;
  private final ProbeErrorClassifier errorClassifier;
  private final RestTemplate ordsRestTemplate;
  private final ObjectMapper objectMapper;
  private final SettingService settingService;
  private final JdbcTemplate jdbcTemplate;
  private final Clock clock;

  public OrdsProbeService(
      BearerTokenService tokenService,
      ProtectedObjectService protectedObjectService,
      AuditService auditService,
      ProbeErrorClassifier errorClassifier,
      @Qualifier("ordsRestTemplate") RestTemplate ordsRestTemplate,
      ObjectMapper objectMapper,
      SettingService settingService,
      JdbcTemplate jdbcTemplate,
      Clock clock
  ) {
    this.tokenService = tokenService;
    this.protectedObjectService = protectedObjectService;
    this.auditService = auditService;
    this.errorClassifier = errorClassifier;
    this.ordsRestTemplate = ordsRestTemplate;
    this.objectMapper = objectMapper;
    this.settingService = settingService;
    this.jdbcTemplate = jdbcTemplate;
    this.clock = clock;
  }

  public ProbeResult runProbe(ProbeCommand command) {
    String ordsBaseUrl = settingService.ordsBaseUrl();
    if (ordsBaseUrl == null || ordsBaseUrl.isBlank()) {
      return auditAndReturn(command, ProbeResult.blocked(
          ProbeStatus.ORDS_NOT_CONFIGURED,
          ProbeStatus.ORDS_NOT_CONFIGURED.name(),
          "ORDS base URL이 설정되지 않았습니다. 실제 ORDS 도메인을 BACKOFFICE_ORDS_BASE_URL에 설정한 뒤 백오피스를 재시작하세요."
      ));
    }

    BearerTokenRecord token;
    if (command.tokenKeyId() == null) {
      token = tokenService.findByPlainToken(command.bearerToken());
      if (token == null) {
        return auditAndReturn(command, ProbeResult.blocked(
            ProbeStatus.TOKEN_NOT_FOUND,
            "TOKEN_NOT_FOUND",
            "현재 DB에 등록된 토큰이 아닙니다. 토큰 화면에서 이 환경의 새 토큰을 발급하세요."
        ));
      }
    } else {
      token = tokenService.findById(command.tokenKeyId());
      if (token == null) {
        return auditAndReturn(command, ProbeResult.blocked(
            ProbeStatus.INVALID_TOKEN, "INVALID_TOKEN", "선택한 등록 토큰을 찾을 수 없습니다."));
      }
      if (!tokenService.matches(token, command.bearerToken())) {
        return auditAndReturn(command, ProbeResult.blocked(
            ProbeStatus.INVALID_TOKEN, "INVALID_TOKEN", "입력한 Bearer Token 원문이 선택한 등록 토큰과 일치하지 않습니다."));
      }
    }
    if (!token.active(LocalDateTime.now(clock.withZone(ZoneId.systemDefault())))) {
      return auditAndReturn(command, ProbeResult.blocked(
          ProbeStatus.TOKEN_INACTIVE, "TOKEN_INACTIVE", "만료되었거나 회수된 토큰입니다."));
    }
    ProtectedObject object;
    try {
      object = protectedObjectService.assertEnabled(command.objectId());
    } catch (AppException e) {
      return auditAndReturn(command, ProbeResult.blocked(
          ProbeStatus.OBJECT_DISABLED, "OBJECT_DISABLED", e.getMessage()));
    }

    String requestHeaders = null;
    String requestPayload = null;
    try {
      URI uri = buildUri(ordsBaseUrl, object.ordsPath(), command.limit());
      HttpHeaders headers = new HttpHeaders();
      headers.setBearerAuth(command.bearerToken());
      headers.setContentType(MediaType.APPLICATION_JSON);
      String probeRequestId = UUID.randomUUID().toString();
      headers.set("X-VPD-Probe-Id", probeRequestId);
      String requestBody = command.requestBody() == null || command.requestBody().isBlank()
          ? "{}"
          : command.requestBody().trim();
      requestHeaders = prettyHeaders(maskedRequestHeaders(headers));
      requestPayload = prettyJson(requestBody);
      ResponseEntity<String> response = ordsRestTemplate.exchange(
          uri, HttpMethod.POST, new HttpEntity<>(requestBody, headers), String.class);
      ProbeResult result = parseSuccess(
          response.getBody(),
          object.objectId(),
          requestHeaders,
          requestPayload,
          prettyHeaders(response.getHeaders()),
          prettyJson(response.getBody())
      );
      result = attachFgaExecutionEvidence(result, object, probeRequestId);
      if (isVectorSearchObject(object)) {
        result = addVectorSqlTrace(result, command.bearerToken(), object);
      } else if (!result.hasSqlTrace()) {
        result = addLocalSqlTrace(result, command.bearerToken(), object);
      }
      return auditAndReturn(command, result);
    } catch (HttpStatusCodeException e) {
      ProbeStatus status = errorClassifier.classify(e.getStatusCode(), e.getResponseBodyAsString());
      return auditAndReturn(command, ProbeResult.blocked(
          status,
          status.name(),
          httpErrorMessage(status, e.getResponseBodyAsString()),
          requestHeaders,
          requestPayload,
          prettyHeaders(e.getResponseHeaders()),
          prettyJson(e.getResponseBodyAsString())
      ));
    } catch (ResourceAccessException e) {
      ProbeStatus status = classifyResourceAccess(e);
      return auditAndReturn(command, ProbeResult.blocked(
          status,
          status.name(),
          resourceAccessMessage(status, e),
          requestHeaders,
          requestPayload,
          "{}",
          ""
      ));
    } catch (Exception e) {
      return auditAndReturn(command, ProbeResult.blocked(
          ProbeStatus.INVALID_ORDS_RESPONSE, "INVALID_ORDS_RESPONSE", e.getMessage()));
    }
  }

  private URI buildUri(String baseUrl, String ordsPath, int limit) {
    String path = ordsPath.startsWith("/") ? ordsPath.substring(1) : ordsPath;
    return UriComponentsBuilder.fromUriString(baseUrl)
        .path("/")
        .path(path)
        .queryParam("limit", limit)
        .build()
        .toUri();
  }

  private ProbeResult parseSuccess(
      String body,
      long objectId,
      String requestHeaders,
      String requestPayload,
      String responseHeaders,
      String responseBody
  ) throws Exception {
    JsonNode root = objectMapper.readTree(body);
    JsonNode rowsNode = root.has("rows") ? root.get("rows") : root;
    rowsNode = root.has("items") ? root.get("items") : rowsNode;
    if (!rowsNode.isArray()) {
      throw new AppException("ORDS 응답에 rows 배열이 없습니다.");
    }

    List<Map<String, Object>> rows = new ArrayList<>();
    Set<String> columns = new LinkedHashSet<>();
    for (JsonNode rowNode : rowsNode) {
      Map<String, Object> row = objectMapper.convertValue(rowNode, MAP_TYPE);
      rows.add(row);
      columns.addAll(row.keySet());
    }

    ProbeStatus status = rows.isEmpty() ? ProbeStatus.VPD_DENY_EMPTY_RESULT : ProbeStatus.SUCCESS;
    return new ProbeResult(
        status,
        List.copyOf(columns),
        rows,
        rows.size(),
        findMaskedColumns(objectId, rows),
        null,
        null,
        requestHeaders,
        requestPayload,
        responseHeaders,
        responseBody,
        traceValue(root, "vpd_predicate"),
        traceValue(root, "effective_sql")
    );
  }

  private String traceValue(JsonNode root, String fieldName) {
    JsonNode value = root == null ? null : root.get(fieldName);
    if (value == null || value.isNull() || !value.isValueNode()) {
      return null;
    }
    String text = value.asText();
    return text == null || text.isBlank() ? null : text;
  }

  /**
   * FGA records are durable database audit evidence. Unlike the optional
   * V$SQL diagnostic view, SQL_TEXT and RLS_INFO originate from the SELECT
   * that Oracle actually audited.
   */
  private ProbeResult attachFgaExecutionEvidence(
      ProbeResult result,
      ProtectedObject object,
      String probeRequestId
  ) {
    String owner = object.owner().toUpperCase(Locale.ROOT);
    String name = object.objectName().toUpperCase(Locale.ROOT);
    FgaExecutionEvidence evidence = findUnifiedFgaEvidence(owner, name, probeRequestId);
    if (evidence == null) {
      evidence = findTraditionalFgaEvidence(owner, name, probeRequestId);
    }
    if (evidence == null) {
      return result.withFgaExecutionEvidence(null,
          "이 요청의 DB 감사 행을 찾지 못했습니다. FGA 실행 감사 정책과 ORDS Handler의 X-VPD-Probe-Id 연동을 적용한 뒤 다시 검증하세요.");
    }
    String message = evidence.hasRlsInfo() ? null
        : "FGA 감사 SQL은 기록됐지만 RLS_INFO가 비어 있습니다. 감사 정책과 DB audit trail 설정을 확인하세요.";
    return result.withFgaExecutionEvidence(evidence, message);
  }

  private FgaExecutionEvidence findUnifiedFgaEvidence(
      String owner,
      String name,
      String probeRequestId
  ) {
    return queryFgaEvidence("""
        SELECT event_timestamp_utc AS event_at,
               dbusername AS db_user,
               client_identifier AS client_id,
               action_name AS statement_type,
               sql_text,
               rls_info
        FROM (
          SELECT event_timestamp_utc,
                 dbusername,
                 client_identifier,
                 action_name,
                 sql_text,
                 rls_info
          FROM unified_audit_trail
          WHERE object_schema = ?
            AND object_name = ?
            AND action_name = 'SELECT'
            AND client_identifier = ?
            AND fga_policy_name IS NOT NULL
          ORDER BY event_timestamp_utc DESC
        )
        WHERE ROWNUM = 1
        """, owner, name, probeRequestId);
  }

  private FgaExecutionEvidence findTraditionalFgaEvidence(
      String owner,
      String name,
      String probeRequestId
  ) {
    return queryFgaEvidence("""
        SELECT extended_timestamp AS event_at,
               db_user,
               client_id,
               statement_type,
               sql_text,
               rls_info
        FROM (
          SELECT extended_timestamp,
                 db_user,
                 client_id,
                 statement_type,
                 sql_text,
                 rls_info
          FROM dba_fga_audit_trail
          WHERE object_schema = ?
            AND object_name = ?
            AND statement_type = 'SELECT'
            AND client_id = ?
          ORDER BY extended_timestamp DESC
        )
        WHERE ROWNUM = 1
        """, owner, name, probeRequestId);
  }

  private FgaExecutionEvidence queryFgaEvidence(String sql, Object... arguments) {
    try {
      return jdbcTemplate.query(sql, resultSet -> {
        if (!resultSet.next()) {
          return null;
        }
        return new FgaExecutionEvidence(
            resultSet.getTimestamp("event_at") == null
                ? null : resultSet.getTimestamp("event_at").toInstant().toString(),
            resultSet.getString("db_user"),
            resultSet.getString("client_id"),
            resultSet.getString("statement_type"),
            resultSet.getString("sql_text"),
            resultSet.getString("rls_info")
        );
      }, arguments);
    } catch (RuntimeException exception) {
      log.debug("FGA audit trail query is unavailable: {}", exception.getMessage());
      return null;
    }
  }

  /**
   * V$SQL keeps the statement submitted by ORDS, not Oracle's internally
   * rewritten VPD text. DBMS_XPLAN is therefore the authoritative place to
   * show the predicate that the matching cursor applied.
   */
  private ProbeResult attachRecentExecutionEvidence(ProbeResult result, ProtectedObject object) {
    try {
      ExecutionEvidenceLookup lookup = findRecentExecutionEvidence(object);
      SqlExecutionEvidence evidence = lookup.evidence();
      if (evidence == null) {
        if (lookup.sqlCatalogAccessUnavailable()) {
          return result.withExecutionEvidence(null,
              "백오피스 DB 연결이 실제 DB cursor SQL을 읽지 못했습니다. 현재 권한 검증 결과는 정상입니다. "
                  + "V$SQL과 DBMS_XPLAN은 일반 데이터 접근과 별도인 SYS 진단 권한입니다.",
              lookup.candidates());
        }
        return result.withExecutionEvidence(null,
            "최신 실행 cursor 10건에서 이 보호 대상을 찾지 못했습니다. "
                + "아래 원문 SQL을 직접 확인하세요.",
            lookup.candidates());
      }
      String message = evidence.hasPredicatePlan() ? null
          : "SQL_ID는 찾았지만 DBMS_XPLAN Predicate Information을 읽지 못했습니다. "
              + "백오피스 DB 계정에 V$SQL/DBMS_XPLAN 조회 권한이 필요합니다.";
      return result.withExecutionEvidence(evidence, message, lookup.candidates());
    } catch (RuntimeException exception) {
      log.debug("Recent SQL execution evidence unavailable for {}.{}: {}",
          object.owner(), object.objectName(), exception.getMessage());
      return result.withExecutionEvidence(null,
          "DB 실행 증적을 읽지 못했습니다. 백오피스 DB 계정에 V$SQL과 DBMS_XPLAN 조회 권한이 필요합니다.",
          List.of());
    }
  }

  private ExecutionEvidenceLookup findRecentExecutionEvidence(ProtectedObject object) {
    try {
      return findRecentExecutionEvidence(jdbcTemplate, object);
    } catch (RuntimeException exception) {
      log.debug("Backoffice cursor evidence is unavailable: {}", exception.getMessage());
      return new ExecutionEvidenceLookup(null, List.of(), isSqlCatalogAccessUnavailable(exception));
    }
  }

  private boolean isSqlCatalogAccessUnavailable(RuntimeException exception) {
    String message = exception.getMessage();
    if (message == null) {
      return false;
    }
    String normalized = message.toUpperCase(Locale.ROOT);
    return normalized.contains("ORA-00942")
        || normalized.contains("V_$SQL")
        || normalized.contains("V$SQL");
  }

  private ExecutionEvidenceLookup findRecentExecutionEvidence(
      JdbcTemplate evidenceJdbcTemplate,
      ProtectedObject object
  ) {
    List<RecentSqlCursor> candidates = evidenceJdbcTemplate.query("""
        SELECT sql_id,
               child_number,
               parsing_schema_name,
               sql_fulltext,
               last_active_time,
               executions,
               rows_processed,
               elapsed_time,
               buffer_gets
        FROM (
          SELECT sql_id,
                 child_number,
                 parsing_schema_name,
                 sql_fulltext,
                 last_active_time,
                 executions,
                 rows_processed,
                 elapsed_time,
                 buffer_gets
          FROM v$sql
          WHERE last_active_time IS NOT NULL
          ORDER BY last_active_time DESC
        )
        WHERE ROWNUM <= 10
        """, (resultSet, rowNum) -> new RecentSqlCursor(
        resultSet.getString("sql_id"),
        resultSet.getInt("child_number"),
        resultSet.getString("parsing_schema_name"),
        resultSet.getString("sql_fulltext"),
        resultSet.getTimestamp("last_active_time") == null
            ? null : resultSet.getTimestamp("last_active_time").toLocalDateTime().toString(),
        resultSet.getLong("executions"),
        resultSet.getLong("rows_processed"),
        Math.round(resultSet.getLong("elapsed_time") / 1000.0d),
        resultSet.getLong("buffer_gets")
    ));

    List<SqlExecutionCandidate> candidateViews = candidates.stream()
        .map(candidate -> new SqlExecutionCandidate(
            candidate.sqlId(),
            candidate.childNumber(),
            candidate.parsingSchema(),
            candidate.sqlFulltext(),
            candidate.lastActiveAt(),
            referencesProtectedObject(candidate.sqlFulltext(), object)
        ))
        .toList();

    SqlExecutionEvidence evidence = candidates.stream()
        .filter(candidate -> referencesProtectedObject(candidate.sqlFulltext(), object))
        .findFirst()
        .map(candidate -> new SqlExecutionEvidence(
            candidate.sqlId(),
            candidate.childNumber(),
            candidate.sqlFulltext(),
            candidate.lastActiveAt(),
            candidate.executions(),
            candidate.rowsProcessed(),
            candidate.elapsedMillis(),
            candidate.bufferGets(),
            findPredicatePlan(evidenceJdbcTemplate, candidate.sqlId(), candidate.childNumber())
        ))
        .orElse(null);
    return new ExecutionEvidenceLookup(evidence, candidateViews, false);
  }

  static boolean referencesProtectedObject(String sqlFulltext, ProtectedObject object) {
    if (sqlFulltext == null || object == null) {
      return false;
    }
    String normalizedSql = sqlFulltext.toUpperCase(Locale.ROOT)
        .replace("\"", "")
        .replaceAll("\\s+", "");
    String objectName = (object.owner() + "." + object.objectName())
        .toUpperCase(Locale.ROOT);
    return normalizedSql.contains(objectName);
  }

  private String findPredicatePlan(
      JdbcTemplate evidenceJdbcTemplate,
      String sqlId,
      int childNumber
  ) {
    try {
      List<String> lines = evidenceJdbcTemplate.query("""
          SELECT plan_table_output
          FROM TABLE(DBMS_XPLAN.DISPLAY_CURSOR(?, ?, 'ALLSTATS LAST +PREDICATE +ALIAS'))
          """, (resultSet, rowNum) -> resultSet.getString(1), sqlId, childNumber);
      return lines.isEmpty() ? null : String.join(System.lineSeparator(), lines);
    } catch (RuntimeException exception) {
      log.debug("DBMS_XPLAN is unavailable for SQL_ID {} child {}: {}",
          sqlId, childNumber, exception.getMessage());
      return null;
    }
  }

  private ProbeResult addLocalSqlTrace(ProbeResult result, String bearerToken, ProtectedObject object) {
    String predicate = findVpdPredicate(bearerToken, object);
    if (predicate == null || predicate.isBlank()) {
      return result;
    }
    List<String> columns;
    try {
      columns = protectedObjectService.findColumns(object.objectId()).stream()
          .map(column -> "o." + column.columnName())
          .toList();
    } catch (RuntimeException ignored) {
      return result;
    }
    if (columns.isEmpty()) {
      return result;
    }
    String effectiveSql = "SELECT " + String.join(", ", columns)
        + " FROM " + object.owner() + "." + object.objectName() + " o"
        + " WHERE (" + predicate + ")"
        + " AND ROWNUM <= LEAST(GREATEST(NVL(:row_limit, 50), 1), 500)";
    return result.withSqlTrace(predicate, effectiveSql);
  }

  private ProbeResult addVectorSqlTrace(ProbeResult result, String bearerToken, ProtectedObject object) {
    String predicate = findVpdPredicate(bearerToken, object);
    if (predicate == null || predicate.isBlank()) {
      return result;
    }
    return result.withSqlTrace(predicate, vectorEffectiveSql(object, predicate));
  }

  static String vectorEffectiveSql(ProtectedObject object, String predicate) {
    String vpdFilter = isAllowAllPredicate(predicate) ? "" : " AND (" + predicate + ")";
    return "SELECT chunk_id, document_id, chunk_no, title, chunk_text, source_uri, tech_tag, score"
        + " FROM (SELECT d.chunk_id, d.document_id, d.chunk_no, d.title, d.chunk_text,"
        + " d.source_uri, d.tech_tag, VECTOR_DISTANCE(d.embedding, TO_VECTOR(:embedding), COSINE) AS score"
        + " FROM " + object.owner() + "." + object.objectName() + " d"
        + " WHERE d.embedding IS NOT NULL" + vpdFilter
        + " ORDER BY score) ranked_chunks"
        + " WHERE ROWNUM <= LEAST(GREATEST(NVL(:row_limit, 10), 1), 100)";
  }

  private static boolean isAllowAllPredicate(String predicate) {
    return predicate != null && predicate.trim().replaceAll("\\s+", " ").equals("1 = 1");
  }

  private String findVpdPredicate(String bearerToken, ProtectedObject object) {
    try {
      return jdbcTemplate.execute((ConnectionCallback<String>) connection -> {
        try {
          executeContextSetter(connection, bearerToken);
          try (PreparedStatement statement = connection.prepareStatement(
              "SELECT admin.cb_agent_doc_vpd_filter(?, ?) FROM dual")) {
            statement.setString(1, object.owner());
            statement.setString(2, object.objectName());
            try (ResultSet result = statement.executeQuery()) {
              return result.next() ? result.getString(1) : null;
            }
          }
        } finally {
          clearContext(connection);
        }
      });
    } catch (RuntimeException exception) {
      // The backoffice may use a different DB account or a database without
      // the optional trace privilege. The ORDS response remains authoritative
      // when the Handler itself returned trace fields.
      log.debug("VPD SQL trace unavailable for {}.{}: {}",
          object.owner(), object.objectName(), exception.getMessage());
      return null;
    }
  }

  private void executeContextSetter(Connection connection, String bearerToken) throws java.sql.SQLException {
    try (PreparedStatement statement = connection.prepareStatement(
        "BEGIN admin.cb_agent_ctx_pkg.set_user_by_bearer(?); END;")) {
      statement.setString(1, bearerToken);
      statement.execute();
    }
  }

  private void clearContext(Connection connection) {
    try (PreparedStatement statement = connection.prepareStatement(
        "BEGIN admin.cb_agent_ctx_pkg.clear_user; END;")) {
      statement.execute();
    } catch (java.sql.SQLException exception) {
      // Do not replace a successful ORDS result with a diagnostic cleanup error.
      log.debug("VPD context cleanup failed: {}", exception.getMessage());
    }
  }

  private boolean isVectorSearchObject(ProtectedObject object) {
    return "CB_VECTOR_SEARCH_DOCUMENTS".equalsIgnoreCase(object.objectName());
  }

  private List<String> findMaskedColumns(long objectId, List<Map<String, Object>> rows) {
    if (rows.isEmpty()) {
      return List.of();
    }
    List<ProtectedColumn> sensitiveColumns = protectedObjectService.findColumns(objectId).stream()
        .filter(ProtectedColumn::sensitive)
        .toList();
    if (sensitiveColumns.isEmpty()) {
      return List.of();
    }

    List<String> masked = new ArrayList<>();
    for (ProtectedColumn column : sensitiveColumns) {
      boolean allNull = true;
      for (Map<String, Object> row : rows) {
        for (Map.Entry<String, Object> entry : row.entrySet()) {
          if (entry.getKey().equalsIgnoreCase(column.columnName())) {
            allNull = allNull && entry.getValue() == null;
          }
        }
      }
      // APEX_JSON omits a ref-cursor column whose value is NULL.  The
      // object handlers always select every registered protected column, so
      // a sensitive column missing from every JSON row is also a NULL result
      // and must be reported as masked to the MCP/UI caller.
      if (allNull) {
        masked.add(column.columnName().toLowerCase(Locale.ROOT) + " [" + column.policyLabel() + "]");
      }
    }
    return masked;
  }

  private ProbeResult auditAndReturn(ProbeCommand command, ProbeResult result) {
    auditService.record(new AuditEvent(
        "ORDS_PROBE",
        tokenKeyId(command),
        command.objectId(),
        result.status().name(),
        result.rowCount(),
        result.errorCode(),
        result.errorMessage()
    ));
    return result;
  }

  private Long tokenKeyId(ProbeCommand command) {
    if (command.tokenKeyId() != null) {
      return command.tokenKeyId();
    }
    BearerTokenRecord token = tokenService.findByPlainToken(command.bearerToken());
    return token == null ? null : token.keyId();
  }

  private ProbeStatus classifyResourceAccess(ResourceAccessException exception) {
    if (errorClassifier.isTimeout(exception)) {
      return ProbeStatus.ORDS_TIMEOUT;
    }
    if (errorClassifier.isUnavailable(exception)) {
      return ProbeStatus.ORDS_UNAVAILABLE;
    }
    return ProbeStatus.UNKNOWN_ERROR;
  }

  private String resourceAccessMessage(ProbeStatus status, ResourceAccessException exception) {
    String detail = trimMessage(exception.getMessage());
    if (status == ProbeStatus.ORDS_UNAVAILABLE) {
      return "ORDS 서버에 연결할 수 없습니다. BACKOFFICE_ORDS_BASE_URL의 실제 ORDS 도메인, ORDS 실행 상태, 네트워크 접근을 확인하세요. 상세: "
          + detail;
    }
    if (status == ProbeStatus.ORDS_TIMEOUT) {
      return "ORDS 응답 시간이 초과되었습니다. ORDS 상태와 BACKOFFICE_ORDS_TIMEOUT_SECONDS 설정을 확인하세요. 상세: "
          + detail;
    }
    return detail;
  }

  private String httpErrorMessage(ProbeStatus status, String body) {
    String detail = trimMessage(body);
    if (status == ProbeStatus.ORDS_PATH_NOT_FOUND) {
      return "ORDS 경로를 찾을 수 없습니다. 설정의 Base URL과 보호 객체 ORDS Path가 실제 ORDS schema mapping/module/template 경로와 일치하는지 확인하세요. 상세: "
          + detail;
    }
    return detail;
  }

  private String trimMessage(String body) {
    if (body == null) {
      return null;
    }
    return body.length() <= 500 ? body : body.substring(0, 500);
  }

  private HttpHeaders maskedRequestHeaders(HttpHeaders headers) {
    HttpHeaders masked = new HttpHeaders();
    masked.putAll(headers);
    List<String> authorization = headers.get(HttpHeaders.AUTHORIZATION);
    if (authorization != null && !authorization.isEmpty()) {
      masked.set(HttpHeaders.AUTHORIZATION, maskBearer(authorization.get(0)));
    }
    return masked;
  }

  private String maskBearer(String value) {
    if (value == null || value.isBlank()) {
      return "";
    }
    if (!value.toLowerCase(Locale.ROOT).startsWith("bearer ")) {
      return "****";
    }
    String token = value.substring("Bearer ".length());
    String suffix = token.length() <= 6 ? "" : token.substring(token.length() - 6);
    return "Bearer ****" + suffix;
  }

  private String prettyHeaders(HttpHeaders headers) {
    if (headers == null || headers.isEmpty()) {
      return "{}";
    }
    return prettyObject(headers);
  }

  private String prettyJson(String body) {
    if (body == null || body.isBlank()) {
      return "";
    }
    try {
      return objectMapper.writerWithDefaultPrettyPrinter()
          .writeValueAsString(objectMapper.readTree(body));
    } catch (Exception ignored) {
      return body;
    }
  }

  private String prettyObject(Object value) {
    try {
      return objectMapper.writerWithDefaultPrettyPrinter().writeValueAsString(value);
    } catch (Exception ignored) {
      return String.valueOf(value);
    }
  }
}
