agent-foundation/src/main/java/com/company/agentfoundation/integration/ngp/NgpTokenProvider.java

package com.company.agentfoundation.integration.ngp;

import com.company.agentfoundation.config.FoundationProperties;
import com.fasterxml.jackson.databind.JsonNode;
import io.netty.channel.ChannelOption;
import org.springframework.http.MediaType;
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.netty.http.client.HttpClient;

import java.time.Duration;
import java.util.Objects;

@Component
public class NgpTokenProvider {
  private final FoundationProperties.Ngp config;
  private final NgpTokenRepository tokenRepository;
  private final WebClient webClient;
  private final Duration readTimeout;
  private volatile String accessToken;
  private volatile String refreshToken;

  public NgpTokenProvider(
      FoundationProperties properties,
      NgpTokenRepository tokenRepository,
      WebClient.Builder builder
  ) {
    this.config = properties.getNgp();
    this.tokenRepository = tokenRepository;
    HttpClient httpClient = HttpClient.create()
        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, Math.max(config.getConnectTimeoutMs(), 1))
        .responseTimeout(Duration.ofMillis(Math.max(config.getReadTimeoutMs(), 1)));
    this.webClient = builder.clone()
        .baseUrl(config.getBaseUrl())
        .clientConnector(new ReactorClientHttpConnector(httpClient))
        .codecs(codecs -> codecs.defaultCodecs()
            .maxInMemorySize(Math.max(config.getMaxResponseBytes(), 256 * 1024)))
        .build();
    this.readTimeout = Duration.ofMillis(Math.max(config.getReadTimeoutMs(), 1));
    NgpTokenRepository.TokenState persisted = tokenRepository.find(config.getAppKey()).orElse(null);
    this.accessToken = persisted == null || blank(persisted.accessToken())
        ? config.getAccessToken() : persisted.accessToken();
    this.refreshToken = persisted == null || blank(persisted.refreshToken())
        ? config.getRefreshToken() : persisted.refreshToken();
  }

  public String getAccessToken() {
    ensureBaseConfiguration();
    if (!blank(accessToken)) return accessToken;
    return refreshAccessToken();
  }

  public synchronized String refreshAccessToken() {
    return refreshAccessTokenInternal();
  }

  /**
   * Refreshes only when the rejected token is still the active token. Concurrent
   * callers that already observed the same expired token reuse the first caller's
   * refreshed value instead of rotating the refresh token repeatedly.
   */
  public synchronized String refreshAccessToken(String rejectedAccessToken) {
    ensureBaseConfiguration();
    if (!blank(accessToken) && !Objects.equals(accessToken, rejectedAccessToken)) return accessToken;
    return refreshAccessTokenInternal();
  }

  private String refreshAccessTokenInternal() {
    ensureBaseConfiguration();
    if (blank(refreshToken)) throw new NgpApiException("NGP access token 已失效，且未配置 refresh token");
    try {
      JsonNode payload = webClient.post()
          .uri(builder -> builder.path(config.getTokenPath())
              .queryParam("app_key", config.getAppKey())
              .queryParam("app_secret", config.getAppSecret())
              .queryParam("grant_type", "refresh_token")
              .queryParam("refresh_token", refreshToken)
              .build())
          .accept(MediaType.APPLICATION_JSON)
          .contentLength(0)
          .retrieve()
          .bodyToMono(JsonNode.class)
          .block(readTimeout);
      JsonNode tokenData = findTokenData(payload);
      if (tokenData == null || blank(tokenData.path("access_token").asText())) {
        throw new NgpApiException("NGP token 刷新失败：" + apiMessage(payload));
      }
      accessToken = tokenData.path("access_token").asText();
      String nextRefreshToken = tokenData.path("refresh_token").asText("");
      if (!nextRefreshToken.isBlank()) refreshToken = nextRefreshToken;
      tokenRepository.save(config.getAppKey(), accessToken, refreshToken);
      return accessToken;
    } catch (NgpApiException exception) {
      throw exception;
    } catch (RuntimeException exception) {
      throw new NgpApiException("NGP token 刷新失败：" + rootMessage(exception), exception);
    }
  }

  private JsonNode findTokenData(JsonNode payload) {
    JsonNode current = payload;
    for (int depth = 0; depth < 4 && current != null && !current.isMissingNode() && !current.isNull(); depth++) {
      if (!blank(current.path("access_token").asText())) return current;
      current = current.path("data");
    }
    return null;
  }

  private void ensureBaseConfiguration() {
    if (!config.isEnabled()) throw new NgpApiException("NGP 接口未启用");
    if (blank(config.getAppKey()) || blank(config.getAppSecret())) {
      throw new NgpApiException("NGP 接口配置不完整：请设置 NGP_APP_KEY 和 NGP_APP_SECRET");
    }
  }

  private String apiMessage(JsonNode payload) {
    if (payload == null || payload.isNull()) return "空响应";
    return payload.path("message").asText(payload.path("msg").asText("未返回 access_token"));
  }

  private boolean blank(String value) {
    return value == null || value.isBlank();
  }

  private String rootMessage(Throwable throwable) {
    Throwable current = throwable;
    while (current.getCause() != null && current.getCause() != current) current = current.getCause();
    return current.getMessage() == null ? current.getClass().getSimpleName() : current.getMessage();
  }
}