content: /*
   * Inkless
   * Copyright (C) 2024 - 2025 Aiven OY
   *
   * This program is free software: you can redistribute it and/or modify
   * it under the terms of the GNU Affero General Public License as published by
   * the Free Software Foundation, either version 3 of the License, or
   * (at your option) any later version.
   *
   * This program is distributed in the hope that it will be useful,
   * but WITHOUT ANY WARRANTY; without even the implied warranty of
   * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   * GNU Affero General Public License for more details.
   *
   * You should have received a copy of the GNU Affero General Public License
   * along with this program.  If not, see <http://www.gnu.org/licenses/>.
   */
  package io.aiven.inkless.control_plane;
  
  import org.apache.kafka.common.TopicIdPartition;
  import org.apache.kafka.common.record.RecordBatch;
  import org.apache.kafka.common.record.TimestampType;
  
  public record CommitBatchRequest(
          byte magic,
          int requestId,
          TopicIdPartition topicIdPartition,
          int byteOffset,
          int size,
          long baseOffset,
          long lastOffset,
          long batchMaxTimestamp,
          TimestampType messageTimestampType,
          long producerId,
          short producerEpoch,
          int baseSequence,
          int lastSequence) {
  
      public static CommitBatchRequest of(
          int requestId,
          TopicIdPartition topicIdPartition,
          int byteOffset,
          RecordBatch batch
      ) {
          return new CommitBatchRequest(
              batch.magic(),
              requestId,
              topicIdPartition,
              byteOffset,
              batch.sizeInBytes(),
              batch.baseOffset(), batch.lastOffset(),
              batch.maxTimestamp(), batch.timestampType(),
              batch.producerId(), batch.producerEpoch(),
              batch.baseSequence(), batch.lastSequence()
          );
      }
  
      // Accessible for testing
      public static CommitBatchRequest of(
          int requestId,
          TopicIdPartition topicIdPartition,
          int byteOffset,
          int size,
          long baseOffset,
          long lastOffset,
          long batchMaxTimestamp,
          TimestampType messageTimestampType
      ) {
          return new CommitBatchRequest(
              RecordBatch.CURRENT_MAGIC_VALUE, requestId, topicIdPartition,
              byteOffset, size, baseOffset, lastOffset, batchMaxTimestamp, messageTimestampType,
              RecordBatch.NO_PRODUCER_ID, RecordBatch.NO_PRODUCER_EPOCH, RecordBatch.NO_SEQUENCE, RecordBatch.NO_SEQUENCE);
      }
  
      // Visible for testing
      public static CommitBatchRequest idempotent(
          int requestId,
          TopicIdPartition topicIdPartition,
          int byteOffset,
          int size,
          long baseOffset,
          long lastOffset,
          long batchMaxTimestamp,
          TimestampType messageTimestampType,
          long producerId,
          short producerEpoch,
          int baseSequence,
          int lastSequence
      ) {
          return new CommitBatchRequest(
              RecordBatch.CURRENT_MAGIC_VALUE, requestId, topicIdPartition,
              byteOffset, size, baseOffset, lastOffset, batchMaxTimestamp, messageTimestampType,
              producerId, producerEpoch, baseSequence, lastSequence);
      }
  
      public boolean hasProducerId() {
          return producerId > RecordBatch.NO_PRODUCER_ID;
      }
  
      public int offsetDelta() {
          return (int) (lastOffset - baseOffset);
      }
  }
  