StreamHTTPClient.java

package io.getstream.services.framework;

import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.fasterxml.jackson.databind.module.SimpleModule;
import com.fasterxml.jackson.databind.util.StdDateFormat;
import io.jsonwebtoken.Jwts;
import io.jsonwebtoken.SignatureAlgorithm;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.security.Key;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.TimeUnit;
import javax.crypto.spec.SecretKeySpec;
import okhttp3.ConnectionPool;
import okhttp3.Dispatcher;
import okhttp3.HttpUrl;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;

public class StreamHTTPClient {
  public static final String API_KEY_PROP_NAME = "io.getstream.apiKey";
  public static final String API_SECRET_PROP_NAME = "io.getstream.apiSecret";
  public static final String API_TIMEOUT_PROP_NAME = "io.getstream.timeout";
  public static final String API_URL_PROP_NAME = "io.getstream.url";
  public static final String API_LOG_LEVEL_PROP_NAME = "io.getstream.debug.logLevel";
  public static final String API_CONNECTION_MAX_AGE_PROP_NAME = "io.getstream.connection.maxAge";
  private static final String API_DEFAULT_URL = "https://chat.stream-io-api.com";
  private static final long DEFAULT_CONNECTION_MAX_AGE_SECONDS = 59;

  @NotNull private final String sdkVersion = readSdkVersion();

  @NotNull
  private final ObjectMapper objectMapper =
      new ObjectMapper()
          .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
          .setSerializationInclusion(JsonInclude.Include.NON_NULL)
          .setDateFormat(
              new StdDateFormat()
                  .withColonInTimeZone(true)
                  .withTimeZone(TimeZone.getTimeZone("UTC")))
          .enable(DeserializationFeature.READ_UNKNOWN_ENUM_VALUES_USING_DEFAULT_VALUE)
          .disable(SerializationFeature.FAIL_ON_EMPTY_BEANS)
          .registerModule(
              new SimpleModule()
                  .addDeserializer(Date.class, new NanosecondTimestampDeserializer()));

  @NotNull private String apiSecret;
  @NotNull private String apiKey;
  private long timeout = 10000;
  private long connectionMaxAgeSeconds = DEFAULT_CONNECTION_MAX_AGE_SECONDS;
  @NotNull private String logLevel = "NONE";
  @NotNull private String baseUrl = API_DEFAULT_URL;
  @NotNull private OkHttpClient client;
  @NotNull private StreamClientOptions options = new StreamClientOptions();

  public StreamHTTPClient(@NotNull String apiKey, @NotNull String apiSecret) {
    setCredetials(apiKey, apiSecret);
    logEffectiveConfig();
  }

  public StreamHTTPClient(
      @NotNull String apiKey, @NotNull String apiSecret, @NotNull OkHttpClient httpClient) {
    this.options = new StreamClientOptions().setHttpClient(httpClient);
    this.apiKey = apiKey;
    this.apiSecret = apiSecret;
    var jwtToken = buildJWT(apiSecret);
    this.client = buildHTTPClient(jwtToken, httpClient.newBuilder());
    logEffectiveConfig();
  }

  public StreamHTTPClient(
      @NotNull String apiKey, @NotNull String apiSecret, @NotNull StreamClientOptions options) {
    this.options = options;
    if (options.hasUserHttpClient()) {
      // Escape hatch: user owns the OkHttpClient. None of the pool/timeout knobs apply.
      this.apiKey = apiKey;
      this.apiSecret = apiSecret;
      var jwtToken = buildJWT(apiSecret);
      this.client = buildHTTPClient(jwtToken, options.getHttpClient().newBuilder());
    } else {
      setCredetials(apiKey, apiSecret);
    }
    logEffectiveConfig();
  }

  // default constructor using ENV or System properties
  // env vars have priority over system properties
  public StreamHTTPClient() {
    this(System.getProperties());
  }

  public StreamHTTPClient(Properties properties) throws IllegalArgumentException {
    this(properties, new StreamClientOptions());
  }

  public StreamHTTPClient(Properties properties, @NotNull StreamClientOptions options)
      throws IllegalArgumentException {
    // Set options before reading env/properties so env overrides (timeout, connection max-age)
    // fold into them and the caller's injected logger is used for the client.initialized event.
    this.options = options;
    readPropertiesAndEnv(properties);

    if (apiKey == null || apiKey.isEmpty()) {
      throw new IllegalArgumentException("apiKey and apiSecret are required");
    }

    if (apiSecret == null || apiSecret.isEmpty()) {
      throw new IllegalArgumentException("apiSecret is required");
    }

    setCredetials(apiKey, apiSecret);
    logEffectiveConfig();
  }

  private static @NotNull String buildJWT(String apiSecret) {
    Key signingKey =
        new SecretKeySpec(
            apiSecret.getBytes(StandardCharsets.UTF_8), SignatureAlgorithm.HS256.getJcaName());
    // We set issued at 5 seconds ago to avoid problems like JWTAuth error in case of clock drift
    GregorianCalendar calendar = new GregorianCalendar();
    calendar.add(Calendar.SECOND, -5);
    return Jwts.builder()
        .issuedAt(new Date())
        .claim("server", true)
        .signWith(signingKey, SignatureAlgorithm.HS256)
        .compact();
  }

  private static @NotNull String readSdkVersion() {
    var clsLoader = StreamHTTPClient.class.getClassLoader();
    try (var inputStream = clsLoader.getResourceAsStream("version.properties")) {
      var properties = new Properties();
      properties.load(inputStream);
      return properties.getProperty("version");
    } catch (IOException ex) {
      throw new IllegalStateException(ex);
    }
  }

  @NotNull
  public OkHttpClient getHttpClient() {
    return client;
  }

  @NotNull
  public ObjectMapper getObjectMapper() {
    return objectMapper;
  }

  @NotNull
  public String getBaseUrl() {
    return baseUrl;
  }

  // Why: construction-time / test-support only (points a client at MockWebServer). Package-private
  // to keep it off the public API; not safe for concurrent post-construction mutation.
  void setBaseUrl(@NotNull String baseUrl) {
    this.baseUrl = baseUrl;
  }

  /** The SLF4J logger for structured events (a no-op logger when none was injected). */
  @NotNull
  public Logger getLogger() {
    return options.getLoggerOrNop();
  }

  public boolean getLogBodies() {
    return options.getLogBodies();
  }

  /** Opt-in auto-retry policy (default: disabled, no retries). */
  @NotNull
  public RetryConfig getRetryConfig() {
    return options.getRetry();
  }

  private void setCredetials(@NotNull String apiKey, @NotNull String apiSecret) {
    this.apiKey = apiKey;
    this.apiSecret = apiSecret;
    var jwtToken = buildJWT(apiSecret);
    this.client = buildHTTPClient(jwtToken, defaultHttpClientBuilder());
  }

  private OkHttpClient.Builder defaultHttpClientBuilder() {
    long idleMillis = options.getIdleTimeout().toMillis();
    // ConnectionPool's first arg is the idle-pool SIZE, not a per-host ceiling. The real per-host
    // concurrency cap is Dispatcher.maxRequestsPerHost (OkHttp default 5), so wire maxConnsPerHost
    // to both: the pool keeps that many idle connections warm, the dispatcher caps in-flight
    // requests per host.
    var dispatcher = new Dispatcher();
    dispatcher.setMaxRequestsPerHost(options.getMaxConnsPerHost());
    return new OkHttpClient.Builder()
        .dispatcher(dispatcher)
        .connectionPool(
            new ConnectionPool(options.getMaxConnsPerHost(), idleMillis, TimeUnit.MILLISECONDS))
        .connectTimeout(options.getConnectTimeout())
        .callTimeout(options.getRequestTimeout());
  }

  private void logEffectiveConfig() {
    Logger logger = getLogger();
    logger.info(
        "client.initialized stream.sdk.name=stream-sdk-java stream.sdk.version={}"
            + " stream.client.max_conns_per_host={} stream.client.idle_timeout_seconds={}"
            + " stream.client.connect_timeout_seconds={} stream.client.request_timeout_seconds={}"
            + " stream.client.gzip_enabled={} stream.client.user_http_client={}"
            + " stream.client.log_bodies={}",
        sdkVersion,
        options.getMaxConnsPerHost(),
        options.getIdleTimeout().toSeconds(),
        options.getConnectTimeout().toSeconds(),
        options.getRequestTimeout().toSeconds(),
        true,
        options.hasUserHttpClient(),
        options.getLogBodies());
    if (options.getLogBodies()) {
      logger.warn(
          "HTTP request/response bodies will be logged. Auth headers and known-secret fields are"
              + " still redacted, but other sensitive data (messages, PII) may appear in logs."
              + " Disable for production.");
    }
  }

  private void readPropertiesAndEnv(Properties properties) {
    var env = System.getenv();

    var propLogLevel = properties.getProperty(API_LOG_LEVEL_PROP_NAME);
    if (propLogLevel != null) {
      this.logLevel = propLogLevel;
    }

    var envApiSecret =
        env.getOrDefault("STREAM_API_SECRET", System.getProperty(API_SECRET_PROP_NAME));
    if (envApiSecret != null) {
      this.apiSecret = envApiSecret;
    }

    var propAPIKey = properties.getProperty(API_KEY_PROP_NAME);
    var envApiKey = env.getOrDefault("STREAM_API_KEY", System.getProperty(API_KEY_PROP_NAME));
    if (envApiKey != null) {
      this.apiKey = envApiKey;
    }

    var envTimeout =
        env.getOrDefault("STREAM_API_TIMEOUT", System.getProperty(API_TIMEOUT_PROP_NAME));
    if (envTimeout != null) {
      timeout = Long.parseLong(envTimeout);
      // Fold the legacy env/property override into the options object so the request-timeout knob
      // actually honors it. Only done when the value was explicitly provided, so an unset env var
      // leaves the StreamClientOptions default (30s) intact rather than the bare 10000ms field.
      options.setRequestTimeout(Duration.ofMillis(timeout));
    }

    var envConnectionMaxAge =
        env.getOrDefault(
            "STREAM_API_CONNECTION_MAX_AGE", System.getProperty(API_CONNECTION_MAX_AGE_PROP_NAME));
    if (envConnectionMaxAge != null) {
      connectionMaxAgeSeconds = Long.parseLong(envConnectionMaxAge);
      // Same as above: an explicit max-age maps onto the idle-timeout knob; absent, the options
      // default (55s) wins over the bare 59s field.
      options.setIdleTimeout(Duration.ofSeconds(connectionMaxAgeSeconds));
    }

    // Treat an empty STREAM_BASE_URL env var as unset (a common CI pattern: `STREAM_BASE_URL:
    // ${{ vars.STREAM_BASE_URL }}` renders to empty when the variable is unset). Empty would
    // otherwise wipe out any io.getstream.url system property set by tests.
    var envBaseUrl = env.get("STREAM_BASE_URL");
    if (envBaseUrl == null || envBaseUrl.isBlank()) {
      envBaseUrl = System.getProperty(API_URL_PROP_NAME);
    }
    if (envBaseUrl != null && !envBaseUrl.isBlank()) {
      this.baseUrl = envBaseUrl;
    }
  }

  @SuppressWarnings("deprecation")
  private @NotNull HttpLoggingInterceptor.Level getLogLevel() {
    return HttpLoggingInterceptor.Level.valueOf(logLevel);
  }

  @SuppressWarnings("deprecation")
  private OkHttpClient buildHTTPClient(String jwtToken, OkHttpClient.Builder httpClient) {
    httpClient.interceptors().clear();

    HttpLoggingInterceptor loggingInterceptor =
        new HttpLoggingInterceptor().setLevel(getLogLevel());
    httpClient.addInterceptor(loggingInterceptor);

    httpClient.addInterceptor(
        chain -> {
          Request original = chain.request();
          HttpUrl url = original.url().newBuilder().addQueryParameter("api_key", apiKey).build();
          Request request =
              original
                  .newBuilder()
                  .url(url)
                  .header("Content-Type", "application/json")
                  .header("X-Stream-Client", "stream-java-client-" + sdkVersion)
                  .header("Stream-Auth-Type", "jwt")
                  .header("Authorization", jwtToken)
                  .build();
          return chain.proceed(request);
        });
    return httpClient.build();
  }

  @NotNull
  public String getApiSecret() {
    return apiSecret;
  }

  @NotNull
  public String getApiKey() {
    return apiKey;
  }
}