package org.apache.kafka.coordinator.group.api.streams;

import org.apache.kafka.common.Configurable;
import org.apache.kafka.server.authorizer.AuthorizableRequestContext;
import java.util.concurrent.CompletableFuture;

/**
 * A broker-side plugin that manages topology descriptions for streams groups.
 *
 * <p>Implementations receive topology descriptions pushed by Kafka Streams clients
 * and can store, forward, or expose them however they see fit.
 *
 * <p>The broker calls {@link #requiresTopologyPush} on every heartbeat to determine
 * whether the client should send its topology description. Implementations that need
 * to consult an external service should kick off that work asynchronously on the
 * first call and return {@code false} until the result is available.
 *
 * <p>Every method takes an {@link AuthorizableRequestContext} as its first argument.
 *
 * <p>Implementations must be thread-safe. {@link #setTopology} may be called
 * concurrently by multiple group members observing
 * {@code TopologyDescriptionRequired=true} in the same heartbeat cycle; concurrent
 * calls with the same {@code (groupId, topologyEpoch)} pair carry identical data
 * and must be treated as idempotent.
 */
public interface StreamsGroupTopologyDescriptionPlugin extends Configurable, AutoCloseable {

    /**
     * Returns whether the broker should request a topology push from the client.
     *
     * <p>Called on every successful heartbeat. Not called for members in
     * {@code STALE_TOPOLOGY} status. If this method returns {@code true}, the broker
     * sets {@code TopologyDescriptionRequired=true} in the heartbeat response.
     *
     * <p>This method should not throw. Failure modes should be handled internally
     * and converted to a return value of {@code false}. The broker defensively
     * catches any exception and treats it as {@code false}, logging at WARN; plugins
     * should not rely on this backstop.
     *
     * <p>This method is on the heartbeat path and must return quickly. Implementations
     * that need to consult an external service should return {@code false} until the
     * result is available.
     *
     * <p>See the KIP's <em>Plugin Implementation Guidelines</em> section for how
     * implementations should handle in-flight tracking, retries, and topology
     * expiration.
     *
     * @param requestContext the context of the heartbeat request
     * @param groupId the streams group ID
     * @param groupCreationTimeMs the timestamp when the group was created. A value
     *                            of {@code 0} means "unset". Plugins should treat {@code 0}
     *                            as "incarnation indistinguishable" and should not cross-check
     *                            stored epoch values against another stored description that
     *                            also had {@code groupCreationTimeMs == 0}.
     * @param topologyEpoch the topology epoch
     * @return true if the broker should request a topology push from the client
     */
    boolean requiresTopologyPush(AuthorizableRequestContext requestContext,
                                 String groupId, long groupCreationTimeMs, int topologyEpoch);


    /**
     * Called when a client sends a topology description for a streams group.
     * This method may be called concurrently by multiple members of the same group;
     * all calls for the same (groupId, topologyEpoch) carry identical data.
     *
     * <p>The returned future completes when the topology has been persisted or
     * forwarded. All failures must be signalled by completing the returned future
     * exceptionally — implementations must not throw synchronously from this method.
     * The broker handles the future's completion exception as follows:
     * {@link org.apache.kafka.common.errors.InvalidRequestException} maps to
     * {@code INVALID_REQUEST}; {@link org.apache.kafka.common.errors.TopologyDescriptionTooLargeException}
     * maps to {@code TOPOLOGY_DESCRIPTION_TOO_LARGE}; any other exception maps to
     * {@code TOPOLOGY_DESCRIPTION_UPDATE_FAILED} and is logged at WARN.
     *
     * @param requestContext the context of the UpdateStreamsGroupTopologyDescription request
     * @param groupId the streams group ID
     * @param groupCreationTimeMs the timestamp when the group was created
     * @param topologyEpoch the topology epoch
     * @param description the topology description
     * @return a future that completes when the operation is done
     */
    CompletableFuture<Void> setTopology(AuthorizableRequestContext requestContext,
                                        String groupId, long groupCreationTimeMs, int topologyEpoch,
                                        StreamsGroupTopologyDescription description);

    /**
     * Called when a group is explicitly deleted via DeleteGroups. Removes any topology
     * description stored for this group.
     *
     * <p>The returned future completes when the deletion has been processed.
     * If it completes exceptionally, the broker logs the error; the outcome does not
     * affect the DeleteGroups response returned to the caller.
     *
     * @param requestContext the context of the DeleteGroups request
     * @param groupId the streams group ID
     * @return a future that completes when the operation is done
     */
    CompletableFuture<Void> deleteTopology(AuthorizableRequestContext requestContext, String groupId);

    /**
     * Called to retrieve the stored topology description for a group. This is invoked
     * by the broker when a client calls StreamsGroupDescribe with
     * {@code IncludeTopologyDescription=true}.
     *
     * <p>Returns a future that resolves to the stored topology description for the
     * given {@code (groupId, groupCreationTimeMs, topologyEpoch)} tuple, or to
     * {@code null} if no topology is stored (e.g. no push has succeeded yet, or the
     * stored description is for a different epoch). If the future completes
     * exceptionally, the plugin signals a read error for this group.
     *
     * @param requestContext the context of the StreamsGroupDescribe request
     * @param groupId the streams group ID
     * @param groupCreationTimeMs the timestamp when the group was created
     * @param topologyEpoch the topology epoch the caller is asking about
     * @return a future resolving to the stored topology description, or null if none
     */
    CompletableFuture<StreamsGroupTopologyDescription>
        getTopology(AuthorizableRequestContext requestContext,
                    String groupId, long groupCreationTimeMs, int topologyEpoch);
}