/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.kafka.raft;

import org.apache.kafka.common.Node;
import org.apache.kafka.common.errors.ApiException;
import org.apache.kafka.common.network.ListenerName;
import org.apache.kafka.raft.errors.BufferAllocationException;
import org.apache.kafka.raft.errors.NotLeaderException;
import org.apache.kafka.server.common.KRaftVersion;
import org.apache.kafka.server.common.OffsetAndEpoch;
import org.apache.kafka.snapshot.SnapshotReader;
import org.apache.kafka.snapshot.SnapshotWriter;

import java.util.List;
import java.util.Optional;
import java.util.OptionalInt;
import java.util.OptionalLong;
import java.util.concurrent.CompletableFuture;

public interface RaftClient<T> extends AutoCloseable {

    interface Listener<T> {
        /**
         * Callback which is invoked for all records committed to the log.
         * It is the responsibility of this implementation to invoke {@link BatchReader#close()}
         * after consuming the reader.
         *
         * Note that there is not a one-to-one correspondence between writes through
         * {@link #prepareAppend(int, List)} and this callback. The Raft implementation is free to
         * batch together the records from multiple append calls provided that batch boundaries are
         * respected. Records specified through {@link #prepareAppend(int, List)} are guaranteed
         * to be a subset of a batch provided by the {@link BatchReader}.
         *
         * @param reader reader instance which must be iterated and closed
         */
        void handleCommit(BatchReader<T> reader);

        /**
         * Callback which is invoked when the Listener needs to load a snapshot.
         * It is the responsibility of this implementation to invoke {@link SnapshotReader#close()}
         * after consuming the reader.
         *
         * When handling this call, the implementation must assume that all previous calls
         * to {@link #handleCommit} contain invalid data.
         *
         * @param reader snapshot reader instance which must be iterated and closed
         */
        void handleLoadSnapshot(SnapshotReader<T> reader);

        /**
         * Called on any change to leadership. This includes both when a leader is elected and
         * when a leader steps down or fails.
         *
         * If this node is the leader, then the notification of leadership will be delayed until
         * the implementation of this interface has caught up to the high-watermark through calls to
         * {@link #handleLoadSnapshot(SnapshotReader)} and {@link #handleCommit(BatchReader)}.
         *
         * If this node is not the leader, then this method will be called as soon as possible. In
         * this case the leader may or may not be known for the current epoch.
         *
         * Subsequent calls to this method will expose a monotonically increasing epoch. For a
         * given epoch the leader may be unknown, {@code leader.leaderId} is {@code OptionalInt#empty},
         * or known {@code leader.leaderId} is {@code OptionalInt#of}. Once a leader is known for
         * a given epoch it will remain the leader for that epoch. In other words, the implementation of
         * method should expect this method will be called at most twice for each epoch. Once if the
         * epoch changed but the leader is not known and once when the leader is known for the current
         * epoch.
         *
         * @param leader the current leader and epoch
         */
        default void handleLeaderChange(LeaderAndEpoch leader) {}

        default void beginShutdown() {}
    }

    /**
     * Register a listener to get commit, snapshot and leader notifications.
     *
     * The implementation of this interface assumes that each call to {@code register} uses
     * a different {@code Listener} instance. If the same instance is used for multiple calls
     * to this method, then only one {@code Listener} will be registered.
     *
     * @param listener the listener to register
     */
    void register(Listener<T> listener);

    /**
     * Unregisters a listener.
     *
     * To distinguish from events that happened before the call to {@code unregister} and a future
     * call to {@code register}, different {@code Listener} instances must be used.
     *
     * If the {@code Listener} provided was never registered then the unregistration is ignored.
     *
     * @param listener the listener to unregister
     */
    void unregister(Listener<T> listener);

    /**
     * Returns the current high watermark, or OptionalLong.empty if it is not known.
     */
    OptionalLong highWatermark();

    /**
     * Return the current {@link LeaderAndEpoch}.
     *
     * @return the current leader and epoch
     */
    LeaderAndEpoch leaderAndEpoch();

    /**
     * Get local nodeId if one is defined. This may be absent when the client is used
     * as an anonymous observer, as in the case of the metadata shell.
     *
     * @return optional node id
     */
    OptionalInt nodeId();

    /**
     * Returns the node information for a given voter id and listener.
     *
     * @param id the id of the voter
     * @param listenerName the name of the listener
     * @return the node information if it exists, otherwise {@code Optional.empty()}
     */
    Optional<Node> voterNode(int id, ListenerName listenerName);

    /**
     * Prepare a list of records to be appended to the log.
     *
     * This method will not write any records to the log. To have the KRaft implementation write
     * records to the log, the {@code schedulePreparedAppend} method must be called. There is no
     * guarantee that appended records will be written to the log and eventually committed. However,
     * it is guaranteed that if any of the records become committed, then all of them will be.
     *
     * If the provided current leader epoch does not match the current epoch, which
     * is possible when the state machine has yet to observe the epoch change, then
     * this method will throw an {@link NotLeaderException} to indicate the leader
     * to resign its leadership. The state machine is expected to discard all
     * uncommitted entries after observing an epoch change.
     *
     * @param epoch the current leader epoch
     * @param records the list of records to append
     * @return the expected offset of the last record
     * @throws org.apache.kafka.common.errors.RecordBatchTooLargeException if the size of the
     *         records is greater than the maximum batch size; if this exception is throw none of
     *         the elements in records were committed
     * @throws NotLeaderException if we are not the current leader or the epoch doesn't match the leader epoch
     * @throws BufferAllocationException we failed to allocate memory for the records
     * @throws IllegalStateException if the number of accumulated batches reaches the maximum
     *         number of batches
     */
    long prepareAppend(int epoch, List<T> records);

    /**
     * Schedule for all of prepared batches to get appended to the log.
     *
     * Any batches previously prepared for append with {@code prepareAppend(int List)} will be
     * scheduled to get appended to the log.
     *
     * @throws NotLeaderException if we are not the current leader
     */
    void schedulePreparedAppend();

    /**
     * Attempt a graceful shutdown of the client. This allows the leader to proactively
     * resign and help a new leader to get elected rather than forcing the remaining
     * voters to wait for the fetch timeout.
     *
     * Note that if the client has hit an unexpected exception which has left it in an
     * indeterminate state, then the call to shutdown should be skipped. However, it
     * is still expected that {@link #close()} will be used to clean up any resources
     * in use.
     *
     * @param timeoutMs How long to wait for graceful completion of pending operations.
     * @return A future which is completed when shutdown completes successfully or the timeout expires.
     */
    CompletableFuture<Void> shutdown(int timeoutMs);

    /**
     * Resign the leadership. The leader will give up its leadership in the passed epoch
     * (if it matches the current epoch), and a new election will be held. Note that nothing
     * prevents this node from being reelected as the leader.
     *
     * Notification of successful resignation can be observed through
     * {@link Listener#handleLeaderChange(LeaderAndEpoch)}.
     *
     * @param epoch the epoch to resign from. If this epoch is smaller than the current epoch, this
     *              call will be ignored.
     *
     * @throws IllegalArgumentException - if the passed epoch is invalid (negative or greater than current) or
     * if the listener is not the leader associated with this epoch.
     */
    void resign(int epoch);

    /**
     * Create a writable snapshot file for a committed offset and epoch.
     *
     * The RaftClient assumes that the snapshot returned will contain the records up to, but not
     * including the committed offset and epoch. If no records have been committed, it is possible
     * to generate an empty snapshot using 0 for both the offset and epoch.
     *
     * See {@link SnapshotWriter} for details on how to use this object. If a snapshot already
     * exists then returns an {@link Optional#empty()}.
     *
     * @param snapshotId The ID of the new snapshot, which includes the (exclusive) last committed offset
     *                   and the last committed epoch.
     * @param lastContainedLogTime The append time of the highest record contained in this snapshot
     * @return a writable snapshot if it doesn't already exist
     * @throws IllegalArgumentException if the committed offset is greater than the high-watermark
     *         or less than the log start offset.
     */
    Optional<SnapshotWriter<T>> createSnapshot(OffsetAndEpoch snapshotId, long lastContainedLogTime);

    /**
     * The snapshot id for the latest snapshot.
     *
     * Returns the snapshot id of the latest snapshot, if it exists. If a snapshot doesn't exist, returns an
     * {@link Optional#empty()}.
     *
     * @return the id of the latest snapshot, if it exists
     */
    Optional<OffsetAndEpoch> latestSnapshotId();

    /**
     * Returns the current end of the log. This method is thread-safe.
     *
     * @return the log end offset, which is one greater than the offset of the last record written,
     *         or 0 if there have not been any records written.
     */
    long logEndOffset();

    /**
     * Returns the latest kraft.version, even if it hasn't been committed durably to Raft.
     *
     * @return the current kraft.version.
     */
    KRaftVersion kraftVersion();

    /**
     * Request that the leader to upgrade the kraft version.
     *
     * @param epoch the current epoch
     * @param version the new kraft version to upgrade to
     * @param validateOnly whether to just validate the change and not persist it
     * @throws ApiException when the upgrade fails to validate
     */
    void upgradeKRaftVersion(
        int epoch,
        KRaftVersion version,
        boolean validateOnly
    );
}
