package org.apache.kafka.streams.state;

/**
 * Iterator interface of {@link Value}.
 * <p>
 * Users must call its {@code close} method explicitly upon completeness to release resources,
 * or use try-with-resources statement (available since JDK7) for this {@link Closeable} class.
 * Note that {@code remove()} is not supported.
 *
 * @param <V> Type of values
 */
public interface ValueIterator<V> extends Iterator<V>, Closeable {

    @Override
    void close();

    /**
     * Peek at the next value without advancing the iterator
     * @return the value that would be returned from the next call to next
     */
    V peekNextValue();
}

package org.apache.kafka.streams.query;

@Evolving
public final class VersionedKeyQuery<K, V> implements Query<ValueIterator<VersionedRecord<V>>> {

  private final K key;
  private final Optional<Long> asOfTimestamp;
  private final Optional<Long> untilTimestamp;

  private VersionedKeyQuery(final K key, Optional<Long> asOfTimestamp, Optional<Long> untilTimestamp);

  /**
   * Interactive key query that returns one record with the specified key
   * * and with the latest timestamp
   * @param key The key to retrieve
   * @param <K> The key type
   * @param <V> The value type
   */
  public static <K, V> VersionedKeyQuery<K, V> withKeyLatestValue(final K key);

/**
   * Interactive key query that returns one record with the specified key
   * * and with the greatest timestamp <= untilTimeStamp
   * @param key The key to retrieve
   * @param untilTimeStamp The upperbound for timestamp
   * @param <K> The key type
   * @param <V> The value type
   */
  public static <K, V> VersionedKeyQuery<K, V> withKeyWithTimestampBound(final K key, long untilTimeStamp);

  /**
   * Interactive key query that returns the records with the specified key
   * * within the specified time range.
   * @param key The key to retrieve
   * @param <K> The key type
   * @param <V> The value type
   */
  public static <K, V> VersionedKeyQuery<K, V> withKeyWithTimestampRange(final K key, long asOfTimestamp, long untilTimestamp);

  /**
   * Interactive key query that returns all the records in the store with the specified key.
   * @param key The key to retrieve
   * @param <K> The key type
   * @param <V> The value type
   */
  public static <K, V> VersionedKeyQuery<K, V> withKeyAllVersions(final K key);

  /**
   * The key that was specified for this query.
   */
  public K getKey();


  /**
   * The lower bound for timestamp of the query, if specified
   */
  public Optional<Long> getAsOfTimestamp();

  /**
   * The upper bound for timestamp of the query, if specified
   */
  public Optional<Long> getUntilTimestamp();
}

package org.apache.kafka.streams.query;
@Evolving
public final class VersionedRangeQuery<K, V> implements Query<KeyValueIterator<K, VersionedRecord<V>>> {
    private final Optional<K> lower;
    private final Optional<K> upper;
    private final Optional<Long> asOfTimestamp;
    private final Optional<Long> untilTimestamp;

    private VersionedRangeQuery(
        final Optional<K> lower,
        final Optional<K> upper,
        final Optional<Long> asOfTimestamp,
        final Optional<Long> untilTimestamp);

    /**
     * Interactive range query using a lower and upper bound to filter the keys returned.
     * * For each key only the record with the latest timestamp is returned.
     * @param lower The key that specifies the lower bound of the range
     * @param upper The key that specifies the upper bound of the range
     * @param <K> The key type
     * @param <V> The value type
     */
    public static <K, V> VersionedRangeQuery<K, V> keyRangeLatestValue(final K lower, final K upper);

    /**
     * Interactive range query using a lower bound to filter the keys returned.
     * * For each key only the record with the latest timestamp is returned.
     * @param lower The key that specifies the lower bound of the range
     * @param <K> The key type
     * @param <V> The value type
     */
    public static <K, V> VersionedRangeQuery<K, V> withLowerBoundLatestValue(final K lower);

    /**
     * Interactive range query using a lower bound to filter the keys returned.
     * * For each key only the record with the latest timestamp is returned.
     * @param upper The key that specifies the upper bound of the range
     * @param <K> The key type
     * @param <V> The value type
     */
    public static <K, V> VersionedRangeQuery<K, V> withUpperBoundLatestValue(final K upper);

    /**
     * Interactive scan query that returns all records in the store.
     * * For each key only the record with the latest timestamp is returned.
     * @param <K> The key type
     * @param <V> The value type
     */
    public static <K, V> VersionedRangeQuery<K, V> withNoBoundLatestValue();

    /**
     * Interactive range query using a lower and upper bound to filter the keys returned.
     * * For each key all the records with the greatest timestamp <= untilTimeStamp are returned.
     * @param lower The key that specifies the lower bound of the range
     * @param upper The key that specifies the upper bound of the range
     * @param untilTimeStamp The upperbound for timestamp
     * @param <K> The key type
     * @param <V> The value type
     */
    public static <K, V> VersionedRangeQuery<K, V> KeyRangeWithTimestampBound(final K lower, final K upper, final long untilTimeStamp);

    /**
     * Interactive range query using a lower bound to filter the keys returned.
     * * For each key all the records with the greatest timestamp <= untilTimeStamp are returned.
     * @param lower The key that specifies the lower bound of the range
     * @param untilTimeStamp The upperbound for timestamp
     * @param <K> The key type
     * @param <V> The value type
     */
    public static <K, V> VersionedRangeQuery<K, V> withLowerBoundWithTimestampBound(final K lower, final long untilTimeStamp);

    /**
     * Interactive range query using an upper and upper bound to filter the keys returned.
     * * For each key all the records with the greatest timestamp <= untilTimeStamp are returned.
     * @param upper The key that specifies the upper bound of the range
     * @param untilTimeStamp The upperbound for timestamp
     * @param <K> The key type
     * @param <V> The value type
     */
    public static <K, V> VersionedRangeQuery<K, V> withUpperBoundWithTimestampBound(final K upper, final long untilTimeStamp);

    /**
     * Interactive scan query that returns all records in the store.
     * * For each key all the records with the greatest timestamp <= untilTimeStamp are returned.
     * @param untilTimeStamp The upperbound for timestamp
     * @param <K> The key type
     * @param <V> The value type
     */
    public static <K, V> VersionedRangeQuery<K, V> withNoBoundWithTimestampBound(final long untilTimeStamp);

    /**
     * Interactive range query using a lower and upper bound to filter the keys returned.
     * * For each key all the records within the specified time range are returned.
     * @param lower The key that specifies the lower bound of the range
     * @param upper The key that specifies the upper bound of the range
     * @param <K> The key type
     * @param <V> The value type
     */
    public static <K, V> VersionedRangeQuery<K, V> keyRangeWithTimestampRange(final K lower, final K upper, final long asOfTimestamp, final long untilTimestamp);

    /**
     * Interactive range query using a lower bound to filter the keys returned.
     * * For each key all the records within the specified time range are returned.
     * @param lower The key that specifies the lower bound of the range
     * @param <K> The key type
     * @param <V> The value type
     */
    public static <K, V> VersionedRangeQuery<K, V> withLowerBoundWithTimestampRange(final K lower, final long asOfTimestamp, final long untilTimestamp);

    /**
     * Interactive range query using an upper bound to filter the keys returned.
     * * For each key all the records within the specified time range are returned.
     * @param upper The key that specifies the upper bound of the range
     * @param <K> The key type
     * @param <V> The value type
     */
    public static <K, V> VersionedRangeQuery<K, V> withUpperBoundWithTimestampRange(final K upper, final long asOfTimestamp, final long untilTimestamp);

    /**
     * Interactive scan query that returns all records in the store.
     * * For each key all the records within the specified time range are returned.
     * @param <K> The key type
     * @param <V> The value type
     */
    public static <K, V> VersionedRangeQuery<K, V> withNoBoundWithTimestampRange(final long asOfTimestamp, final long untilTimestamp);

    /**
     * Interactive range query using a lower and upper bound to filter the keys returned.
     * For each key all values from the oldest till the newest record existing in the state store
     * * are returned
     * @param lower The key that specifies the lower bound of the range
     * @param upper The key that specifies the upper bound of the range
     * @param <K> The key type
     * @param <V> The value type
     */
    public static <K, V> VersionedRangeQuery<K, V> keyRangeAllVersions(final K lower, final K upper);

    /**
     * Interactive range query using a lower bound to filter the keys returned.
     * For each key all values from the oldest till the newest record existing in the state store
     * * are returned
     * @param lower The key that specifies the lower bound of the range
     * @param <K> The key type
     * @param <V> The value type
     */
    public static <K, V> VersionedRangeQuery<K, V> withLowerBoundAllVersions(final K lower);

    /**
     * Interactive range query using an upper bound to filter the keys returned.
     * For each key all values from the oldest till the newest record existing in the state store
     * * are returned
     * @param upper The key that specifies the lower bound of the range
     * @param <K> The key type
     * @param <V> The value type
     */
    public static <K, V> VersionedRangeQuery<K, V> withUpperBoundAllVersions(final K upper);

    /**
     * Interactive scan query that returns all records in the store.
     * For each key all values from the oldest till the newest record existing in the state store
     * * are returned
     * @param <K> The key type
     * @param <V> The value type
     */
    public static <K, V> VersionedRangeQuery<K, V> withNoBoundAllVersions();

    /**
     * The lower bound of the query, if specified.
     */
    public Optional<K> getLowerBound();

    /**
     * The upper bound of the query, if specified
     */
    public Optional<K> getUpperBound();

    /**
     * The upper bound for timestamp of the query, if specified
     */
    public Optional<Long> getAsOfTimestamp();
    /**
     * The upper bound for timestamp of the query, if specified
     */
    public Optional<Long> getUntilTimestamp();
}

final VersionedKeyQuery<Integer, Integer> query = VersionedKeyQuery.withKeyAllVersions(1);

final StateQueryRequest<ValueIterator<VersionedRecord<Integer>>> request =
        inStore("my_store").withQuery(query);

final StateQueryResult<ValueIterator<VersionedRecord<Integer>>> versionedKeyResult = kafkaStreams.query(request);

// Get the results from all partitions.
final Map<Integer, QueryResult<ValueIterator<VersionedRecord<Integer>>>> partitionResults = versionedKeyResult.getPartitionResults();
for (final Entry<Integer, QueryResult<ValueIterator<VersionedRecord<Integer>>>> entry : partitionResults.entrySet()) {
	try (final ValueIterator<VersionedRecord<Integer>> iterator = entry.getValue().getResult()) {
        while (iterator.hasNext()) {
          final VersionedRecord<Integer> record = iterator.next();
          Long timestamp = record.timestamp();
		  Integer value = record.value();	
        }
     }
}  

final VersionedRangeQuery<Integer, Integer> query =
        VersionedRangeQuery.keyRangeWithTimestampRange(1, 2, 1690201149, 1690373949);

final StateQueryRequest<KeyValueIterator<Integer, VersionedRecord<Integer>>> request =
        inStore("my_store").withQuery(query);

final StateQueryResult<KeyValueIterator<Integer, VersionedRecord<Integer>>> versionedRangeResult = kafkaStreams.query(request);

// Get the results from all partitions.
final Map<Integer, QueryResult<KeyValueIterator<Integer, VersionedRecord<Integer>>>> partitionResults = versionedRangeResult.getPartitionResults();
for (final Entry<Integer, QueryResult<KeyValueIterator<Integer, VersionedRecord<Integer>>>> entry : partitionResults.entrySet()) {
	try (final KeyValueIterator<Integer, VersionedRecord<Integer>> iterator = entry.getValue().getResult()) {
        while (iterator.hasNext()) {
          final KeyValue<Integer, VersionedRecord<Integer>> record = iterator.next();
          Integer key = record.key;
          Integer value = record.value.value();
          Long timestamp = record.value.timestamp();
        }
     }
}