content: /*
   * Copyright 2021-2024 Aklivity Inc
   *
   * Licensed under the Aklivity Community License (the "License"); you may not use
   * this file except in compliance with the License.  You may obtain a copy of the
   * License at
   *
   *   https://www.aklivity.io/aklivity-community-license/
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
   * specific language governing permissions and limitations under the License.
   */
  package io.aklivity.zilla.runtime.binding.http.kafka.internal.config;
  
  import java.util.List;
  import java.util.function.Supplier;
  
  import org.agrona.DirectBuffer;
  import org.agrona.concurrent.UnsafeBuffer;
  
  import io.aklivity.zilla.runtime.binding.http.kafka.config.HttpKafkaCorrelationConfig;
  import io.aklivity.zilla.runtime.binding.http.kafka.internal.types.Array32FW;
  import io.aklivity.zilla.runtime.binding.http.kafka.internal.types.HttpHeaderFW;
  import io.aklivity.zilla.runtime.binding.http.kafka.internal.types.KafkaAckMode;
  import io.aklivity.zilla.runtime.binding.http.kafka.internal.types.KafkaAckModeFW;
  import io.aklivity.zilla.runtime.binding.http.kafka.internal.types.KafkaFilterFW;
  import io.aklivity.zilla.runtime.binding.http.kafka.internal.types.KafkaHeaderFW;
  import io.aklivity.zilla.runtime.binding.http.kafka.internal.types.KafkaKeyFW;
  import io.aklivity.zilla.runtime.binding.http.kafka.internal.types.KafkaOffsetFW;
  import io.aklivity.zilla.runtime.binding.http.kafka.internal.types.KafkaOffsetType;
  import io.aklivity.zilla.runtime.binding.http.kafka.internal.types.String16FW;
  import io.aklivity.zilla.runtime.binding.http.kafka.internal.types.String8FW;
  
  public class HttpKafkaWithProduceResult
  {
      private static final HttpKafkaWithProduceAsyncHeaderResult HTTP_STATUS_202 =
              new HttpKafkaWithProduceAsyncHeaderResult(new String8FW(":status"), new String16FW("202"));
      private static final HttpKafkaWithProduceAsyncHeaderResult HTTP_STATUS_204 =
              new HttpKafkaWithProduceAsyncHeaderResult(new String8FW(":status"), new String16FW("204"));
      private static final HttpKafkaWithProduceAsyncHeaderResult HTTP_STATUS_400 =
              new HttpKafkaWithProduceAsyncHeaderResult(new String8FW(":status"), new String16FW("400"));
      private static final HttpKafkaWithProduceAsyncHeaderResult HTTP_CONTENT_LENGTH_0 =
              new HttpKafkaWithProduceAsyncHeaderResult(new String8FW("content-length"), new String16FW("0"));
  
      private static final String8FW HTTP_HEADER_NAME_CONTENT_LENGTH = new String8FW("content-length");
      private static final String8FW HTTP_HEADER_NAME_PREFER = new String8FW("prefer");
      private static final String8FW HTTP_HEADER_NAME_IF_MATCH = new String8FW("if-match");
      private static final String8FW HTTP_HEADER_NAME_STATUS = new String8FW(":status");
  
      private static final String16FW HTTP_HEADER_VALUE_204 = new String16FW("204");
  
      private static final KafkaOffsetFW KAFKA_OFFSET_HISTORICAL =
              new KafkaOffsetFW.Builder()
                  .wrap(new UnsafeBuffer(new byte[32]), 0, 32)
                  .partitionId(-1)
                  .partitionOffset(KafkaOffsetType.HISTORICAL.value())
                  .build();
  
      private final long compositeId;
      private final HttpKafkaCorrelationConfig correlation;
      private final String16FW topic;
      private final KafkaAckMode acks;
      private final Supplier<DirectBuffer> keyRef;
      private final List<HttpKafkaWithProduceOverrideResult> overrides;
      private final String16FW ifMatch;
      private final String16FW replyTo;
      private final List<HttpKafkaWithProduceAsyncHeaderResult> async;
      private final HttpKafkaWithProduceHash hash;
      private final long timeout;
      private final boolean idempotent;
      private final String16FW correlationId;
  
      HttpKafkaWithProduceResult(
          long compositeId,
          HttpKafkaCorrelationConfig correlation,
          String16FW topic,
          KafkaAckMode acks,
          Supplier<DirectBuffer> keyRef,
          List<HttpKafkaWithProduceOverrideResult> overrides,
          String16FW ifMatch,
          String16FW replyTo,
          String16FW correlationId,
          String16FW idempotencyKey,
          List<HttpKafkaWithProduceAsyncHeaderResult> async,
          HttpKafkaWithProduceHash hash,
          long timeout)
      {
          this.compositeId = compositeId;
          this.correlation = correlation;
          this.topic = topic;
          this.acks = acks;
          this.keyRef = keyRef;
          this.overrides = overrides;
          this.ifMatch = ifMatch;
          this.replyTo = replyTo;
          this.correlationId = correlationId;
          this.async = async;
          this.hash = hash;
          this.idempotent = idempotencyKey != null;
          this.timeout = timeout;
      }
  
      public long compositeId()
      {
          return compositeId;
      }
  
      public void updateHash(
          DirectBuffer value)
      {
          hash.updateHash(value);
      }
  
      public void digestHash()
      {
          hash.digestHash();
      }
  
      public String16FW topic()
      {
          return topic;
      }
  
      public void partitions(
          Array32FW.Builder<KafkaOffsetFW.Builder, KafkaOffsetFW> builder)
      {
          builder.item(p -> p.set(KAFKA_OFFSET_HISTORICAL));
      }
  
      public String16FW replyTo()
      {
          return replyTo;
      }
  
      public void acks(
          KafkaAckModeFW.Builder builder)
      {
          builder.set(acks);
      }
  
      public void key(
          KafkaKeyFW.Builder builder)
      {
          final DirectBuffer key = keyRef.get();
          if (key != null)
          {
              builder
                  .length(key.capacity())
                  .value(key, 0, key.capacity());
  
              hash.updateHash(key);
          }
      }
  
      public void headers(
          Array32FW<HttpHeaderFW> headers,
          Array32FW.Builder<KafkaHeaderFW.Builder, KafkaHeaderFW> builder)
      {
          headers.forEach(h -> header(h, builder));
  
          if (replyTo != null)
          {
              builder.item(this::replyTo);
          }
  
          if (overrides != null)
          {
              overrides.forEach(o -> builder.item(o::header));
          }
      }
  
      public void headers(
          HttpHeaderFW contentType,
          Array32FW.Builder<KafkaHeaderFW.Builder, KafkaHeaderFW> builder)
      {
          if (contentType != null)
          {
              builder.item(i ->
                  i.nameLen(contentType.name().length())
                   .name(contentType.name().value(), 0, contentType.name().length())
                   .valueLen(contentType.value().length())
                   .value(contentType.value().value(), 0, contentType.value().length()));
          }
  
          if (replyTo != null)
          {
              builder.item(this::replyTo);
          }
  
          if (overrides != null)
          {
              overrides.forEach(o -> builder.item(o::header));
          }
      }
  
      public void trailers(
          Array32FW.Builder<KafkaHeaderFW.Builder, KafkaHeaderFW> builder)
      {
          if (hash != null)
          {
              builder.item(this::correlationId);
          }
      }
  
      private void header(
          HttpHeaderFW header,
          Array32FW.Builder<KafkaHeaderFW.Builder, KafkaHeaderFW> builder)
      {
          final String8FW name = header.name();
  
          if (!HTTP_HEADER_NAME_CONTENT_LENGTH.equals(name) &&
              !HTTP_HEADER_NAME_PREFER.equals(name))
          {
              final String16FW value = HTTP_HEADER_NAME_IF_MATCH.equals(name) ? ifMatch : header.value();
  
              if (value != null)
              {
                  builder.item(i -> i
                      .nameLen(name.length())
                      .name(name.value(), 0, name.length())
                      .valueLen(value.length())
                      .value(value.value(), 0, value.length()));
  
                  hash.updateHash(name.value());
                  hash.updateHash(value.value());
              }
          }
      }
  
      private void replyTo(
          KafkaHeaderFW.Builder builder)
      {
          builder
              .nameLen(correlation.replyTo.length())
              .name(correlation.replyTo.value(), 0, correlation.replyTo.length())
              .valueLen(replyTo.length())
              .value(replyTo.value(), 0, replyTo.length());
  
          hash.updateHash(correlation.replyTo.value());
          hash.updateHash(replyTo.value());
      }
  
      private void correlationId(
          KafkaHeaderFW.Builder builder)
      {
          final String16FW correlationId = hash.correlationId();
  
          final String16FW name = this.correlationId != null ? this.correlationId : correlation.correlationId;
  
          builder
              .nameLen(name.length())
              .name(name.value(), 0, name.length())
              .valueLen(correlationId.length())
              .value(correlationId.value(), 0, correlationId.length());
      }
  
      public void correlated(
          Array32FW<KafkaHeaderFW> headers,
          Array32FW.Builder<HttpHeaderFW.Builder, HttpHeaderFW> builder,
          int contentLength)
      {
          headers.forEach(h -> correlated(h, builder));
  
          if (!headers.anyMatch(h -> HTTP_HEADER_NAME_CONTENT_LENGTH.value().equals(h.name().value())) &&
               headers.matchFirst(h -> HTTP_HEADER_NAME_STATUS.value().equals(h.name().value()) &&
                                       HTTP_HEADER_VALUE_204.value().equals(h.value().value())) == null)
          {
              builder.item(i -> i
                      .name(HTTP_HEADER_NAME_CONTENT_LENGTH)
                      .value(Integer.toString(contentLength)));
          }
      }
  
      private void correlated(
          KafkaHeaderFW header,
          Array32FW.Builder<HttpHeaderFW.Builder, HttpHeaderFW> builder)
      {
          final DirectBuffer name = header.name().value();
          final DirectBuffer value = header.value().value();
  
          if (!correlation.correlationId.value().equals(name))
          {
              builder.item(i -> i
                      .name(name, 0, name.capacity())
                      .value(value, 0, value.capacity()));
          }
      }
  
      public void async(
          Array32FW.Builder<HttpHeaderFW.Builder, HttpHeaderFW> builder)
      {
          if (async != null)
          {
              builder.item(HTTP_STATUS_202::header);
              builder.item(HTTP_CONTENT_LENGTH_0::header);
              async.forEach(a -> builder.item(a::header));
          }
      }
  
      public void noreply(
          Array32FW.Builder<HttpHeaderFW.Builder, HttpHeaderFW> builder)
      {
          if (replyTo == null)
          {
              builder.item(HTTP_STATUS_204::header);
          }
      }
  
      public void error(
          Array32FW.Builder<HttpHeaderFW.Builder, HttpHeaderFW> builder)
      {
          builder.item(HTTP_STATUS_400::header);
      }
  
      public boolean async()
      {
          return async != null;
      }
  
      public boolean idempotent()
      {
          return idempotent;
      }
  
      public boolean reply()
      {
          return replyTo != null;
      }
  
      public long timeout()
      {
          return timeout;
      }
  
      public void filters(
          Array32FW.Builder<KafkaFilterFW.Builder, KafkaFilterFW> builder)
      {
          final String16FW correlationId = hash.correlationId();
  
          if (correlationId != null)
          {
              builder.item(i -> i
                  .conditionsItem(c -> c
                      .header(h -> h
                          .nameLen(correlation.correlationId.length())
                          .name(correlation.correlationId.value(), 0, correlation.correlationId.length())
                          .valueLen(correlationId.length())
                          .value(correlationId.value(), 0, correlationId.length()))));
          }
      }
  }
  