// Copyright (c) Microsoft Corporation. All rights reserved.
// Licensed under the MIT License.

package com.microsoft.bot.builder;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.microsoft.bot.builder.integration.AdapterIntegration;
import com.microsoft.bot.connector.Async;
import com.microsoft.bot.connector.Channels;
import com.microsoft.bot.connector.ConnectorClient;
import com.microsoft.bot.connector.Conversations;
import com.microsoft.bot.connector.ExecutorFactory;
import com.microsoft.bot.connector.OAuthClient;
import com.microsoft.bot.connector.OAuthClientConfig;
import com.microsoft.bot.connector.authentication.AppCredentials;
import com.microsoft.bot.connector.authentication.AuthenticationConfiguration;
import com.microsoft.bot.connector.authentication.AuthenticationConstants;
import com.microsoft.bot.connector.authentication.ChannelProvider;
import com.microsoft.bot.connector.authentication.ClaimsIdentity;
import com.microsoft.bot.connector.authentication.CredentialProvider;
import com.microsoft.bot.connector.authentication.GovernmentAuthenticationConstants;
import com.microsoft.bot.connector.authentication.JwtTokenValidation;
import com.microsoft.bot.connector.authentication.MicrosoftAppCredentials;
import com.microsoft.bot.connector.authentication.MicrosoftGovernmentAppCredentials;
import com.microsoft.bot.connector.authentication.SimpleCredentialProvider;
import com.microsoft.bot.connector.authentication.SkillValidation;
import com.microsoft.bot.connector.rest.RestConnectorClient;
import com.microsoft.bot.connector.rest.RestOAuthClient;
import com.microsoft.bot.schema.AadResourceUrls;
import com.microsoft.bot.schema.Activity;
import com.microsoft.bot.schema.ActivityEventNames;
import com.microsoft.bot.schema.ActivityTypes;
import com.microsoft.bot.schema.CallerIdConstants;
import com.microsoft.bot.schema.ChannelAccount;
import com.microsoft.bot.schema.ConversationAccount;
import com.microsoft.bot.schema.ConversationParameters;
import com.microsoft.bot.schema.ConversationReference;
import com.microsoft.bot.schema.ConversationsResult;
import com.microsoft.bot.schema.DeliveryModes;
import com.microsoft.bot.schema.ExpectedReplies;
import com.microsoft.bot.schema.ResourceResponse;
import com.microsoft.bot.schema.Serialization;
import com.microsoft.bot.schema.SignInResource;
import com.microsoft.bot.schema.TokenExchangeRequest;
import com.microsoft.bot.schema.TokenExchangeState;
import com.microsoft.bot.schema.TokenResponse;
import com.microsoft.bot.schema.TokenStatus;
import com.microsoft.bot.restclient.retry.RetryStrategy;
import java.util.Collections;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.lang3.StringUtils;

import java.net.HttpURLConnection;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.Base64;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;

/**
 * A bot adapter that can connect a bot to a service endpoint.
 *
 * The bot adapter encapsulates authentication processes and sends activities to
 * and receives activities from the Bot Connector Service. When your bot
 * receives an activity, the adapter creates a context object, passes it to your
 * bot's application logic, and sends responses back to the user's channel.
 * <p>
 * Use {@link #use(Middleware)} to add {@link Middleware} objects to your
 * adapter’s middleware collection. The adapter processes and directs incoming
 * activities in through the bot middleware pipeline to your bot’s logic and
 * then back out again. As each activity flows in and out of the bot, each piece
 * of middleware can inspect or act upon the activity, both before and after the
 * bot logic runs.
 * </p>
 * <p>
 * {@link TurnContext} {@link Activity} {@link Bot} {@link Middleware}
 */
public class BotFrameworkAdapter extends BotAdapter
    implements AdapterIntegration, UserTokenProvider, ConnectorClientBuilder {
    /**
     * Key to store InvokeResponse.
     */
    public static final String INVOKE_RESPONSE_KEY = "BotFrameworkAdapter.InvokeResponse";

    /**
     * Key to store ConnectorClient.
     */
    public static final String CONNECTOR_CLIENT_KEY = "ConnectorClient";

    /**
     * Key to store TeamsConnectorClient. For testing only.
     */
    public static final String TEAMSCONNECTOR_CLIENT_KEY = "TeamsConnectorClient";

    private AppCredentials appCredentials;

    /**
     * The credential provider.
     */
    private final CredentialProvider credentialProvider;

    /**
     * The channel provider.
     */
    private ChannelProvider channelProvider;

    /**
     * The authentication configuration.
     */
    private AuthenticationConfiguration authConfiguration;

    /**
     * Rest RetryStrategy.
     */
    private final RetryStrategy connectorClientRetryStrategy;

    /**
     * AppCredentials dictionary.
     */
    private Map<String, AppCredentials> appCredentialMap = new ConcurrentHashMap<>();

    /**
     * ConnectorClient cache.
     */
    private Map<String, ConnectorClient> connectorClients = new ConcurrentHashMap<>();

    /**
     * OAuthClient cache.
     */
    private Map<String, OAuthClient> oAuthClients = new ConcurrentHashMap<>();

    /**
     * Initializes a new instance of the {@link BotFrameworkAdapter} class, using a
     * credential provider.
     *
     * @param withCredentialProvider The credential provider.
     */
    public BotFrameworkAdapter(CredentialProvider withCredentialProvider) {
        this(withCredentialProvider, null, null, null);
    }

    /**
     * Initializes a new instance of the {@link BotFrameworkAdapter} class, using a
     * credential provider.
     *
     * @param withCredentialProvider The credential provider.
     * @param withChannelProvider    The channel provider.
     * @param withRetryStrategy      Retry policy for retrying HTTP operations.
     * @param withMiddleware         The middleware to initially add to the adapter.
     */
    public BotFrameworkAdapter(
        CredentialProvider withCredentialProvider,
        ChannelProvider withChannelProvider,
        RetryStrategy withRetryStrategy,
        Middleware withMiddleware
    ) {
        this(
            withCredentialProvider,
            new AuthenticationConfiguration(),
            withChannelProvider,
            withRetryStrategy,
            withMiddleware
        );
    }

    /**
     * Initializes a new instance of the {@link BotFrameworkAdapter} class, using a
     * credential provider.
     *
     * @param withCredentialProvider The credential provider.
     * @param withAuthConfig         The authentication configuration.
     * @param withChannelProvider    The channel provider.
     * @param withRetryStrategy      Retry policy for retrying HTTP operations.
     * @param withMiddleware         The middleware to initially add to the adapter.
     */
    public BotFrameworkAdapter(
        CredentialProvider withCredentialProvider,
        AuthenticationConfiguration withAuthConfig,
        ChannelProvider withChannelProvider,
        RetryStrategy withRetryStrategy,
        Middleware withMiddleware
    ) {
        if (withCredentialProvider == null) {
            throw new IllegalArgumentException("CredentialProvider cannot be null");
        }

        if (withAuthConfig == null) {
            throw new IllegalArgumentException("AuthenticationConfiguration cannot be null");
        }

        credentialProvider = withCredentialProvider;
        channelProvider = withChannelProvider;
        connectorClientRetryStrategy = withRetryStrategy;
        authConfiguration = withAuthConfig;

        // Relocate the tenantId field used by MS Teams to a new location (from
        // channelData to conversation)
        // This will only occur on activities from teams that include tenant info in
        // channelData but NOT in
        // conversation, thus should be future friendly. However, once the transition is
        // complete. we can
        // remove this.
        use(new TenantIdWorkaroundForTeamsMiddleware());

        if (withMiddleware != null) {
            use(withMiddleware);
        }
    }

    /**
     * Initializes a new instance of the {@link BotFrameworkAdapter} class, using a
     * credential provider.
     *
     * @param withCredentials     The credentials to use.
     * @param withAuthConfig      The authentication configuration.
     * @param withChannelProvider The channel provider.
     * @param withRetryStrategy   Retry policy for retrying HTTP operations.
     * @param withMiddleware      The middleware to initially add to the adapter.
     */
    public BotFrameworkAdapter(
        AppCredentials withCredentials,
        AuthenticationConfiguration withAuthConfig,
        ChannelProvider withChannelProvider,
        RetryStrategy withRetryStrategy,
        Middleware withMiddleware
    ) {
        if (withCredentials == null) {
            throw new IllegalArgumentException("credentials");
        }
        appCredentials = withCredentials;

        credentialProvider = new SimpleCredentialProvider(withCredentials.getAppId(), null);
        channelProvider = withChannelProvider;
        connectorClientRetryStrategy = withRetryStrategy;

        if (withAuthConfig == null) {
            throw new IllegalArgumentException("authConfig");
        }
        authConfiguration = withAuthConfig;

        // Relocate the tenantId field used by MS Teams to a new location (from
        // channelData to conversation)
        // This will only occur on activities from teams that include tenant info in
        // channelData but NOT in
        // conversation, thus should be future friendly. However, once the transition is
        // complete. we can
        // remove this.
        use(new TenantIdWorkaroundForTeamsMiddleware());

        if (withMiddleware != null) {
            use(withMiddleware);
        }
    }

    /**
     * Sends a proactive message from the bot to a conversation.
     *
     * <p>
     * Call this method to proactively send a message to a conversation. Most
     * channels require a user to initiate a conversation with a bot before the bot
     * can send activities to the user.
     * </p>
     * <p>
     * This overload differers from the Node implementation by requiring the BotId
     * to be passed in. The .Net code allows multiple bots to be hosted in a single
     * adapter which isn't something supported by Node.
     * </p>
     *
     * {@link #processActivity(String, Activity, BotCallbackHandler)}
     * {@link BotAdapter#runPipeline(TurnContext, BotCallbackHandler)}
     *
     * @param botAppId  The application ID of the bot. This is the appId returned by
     *                  Portal registration, and is generally found in the
     *                  "MicrosoftAppId" parameter in appSettings.json.
     * @param reference A reference to the conversation to continue.
     * @param callback  The method to call for the resulting bot turn.
     * @return A task that represents the work queued to execute.
     * @throws IllegalArgumentException botAppId, reference, or callback is null.
     */
    @Override
    public CompletableFuture<Void> continueConversation(
        String botAppId,
        ConversationReference reference,
        BotCallbackHandler callback
    ) {
        if (reference == null) {
            return Async.completeExceptionally(new IllegalArgumentException("reference"));
        }

        if (callback == null) {
            return Async.completeExceptionally(new IllegalArgumentException("callback"));
        }

        botAppId = botAppId == null ? "" : botAppId;

        // Hand craft Claims Identity.
        // Adding claims for both Emulator and Channel.
        HashMap<String, String> claims = new HashMap<String, String>();
        claims.put(AuthenticationConstants.AUDIENCE_CLAIM, botAppId);
        claims.put(AuthenticationConstants.APPID_CLAIM, botAppId);

        ClaimsIdentity claimsIdentity = new ClaimsIdentity("ExternalBearer", claims);
        String audience = getBotFrameworkOAuthScope();

        return continueConversation(claimsIdentity, reference, audience, callback);
    }

    /**
     * Sends a proactive message to a conversation.
     *
     * <p>
     * Call this method to proactively send a message to a conversation. Most
     * channels require a user to initiate a conversation with a bot before the bot
     * can send activities to the user.
     * </p>
     *
     * @param claimsIdentity A ClaimsIdentity reference for the conversation.
     * @param reference      A reference to the conversation to continue.
     * @param callback       The method to call for the result bot turn.
     * @return A task that represents the work queued to execute.
     */
    public CompletableFuture<Void> continueConversation(
        ClaimsIdentity claimsIdentity,
        ConversationReference reference,
        BotCallbackHandler callback
    ) {
        return continueConversation(claimsIdentity, reference, getBotFrameworkOAuthScope(), callback);
    }

    /**
     * Sends a proactive message to a conversation.
     *
     * <p>
     * Call this method to proactively send a message to a conversation. Most
     * channels require a user to initiate a conversation with a bot before the bot
     * can send activities to the user.
     * </p>
     *
     * @param claimsIdentity A ClaimsIdentity reference for the conversation.
     * @param reference      A reference to the conversation to continue.
     * @param audience       A value signifying the recipient of the proactive
     *                       message.
     * @param callback       The method to call for the result bot turn.
     * @return A task that represents the work queued to execute.
     */
    public CompletableFuture<Void> continueConversation(
        ClaimsIdentity claimsIdentity,
        ConversationReference reference,
        String audience,
        BotCallbackHandler callback
    ) {
        if (claimsIdentity == null) {
            return Async.completeExceptionally(new IllegalArgumentException("claimsIdentity"));
        }

        if (reference == null) {
            return Async.completeExceptionally(new IllegalArgumentException("reference"));
        }

        if (callback == null) {
            return Async.completeExceptionally(new IllegalArgumentException("callback"));
        }

        if (StringUtils.isEmpty(audience)) {
            return Async.completeExceptionally(new IllegalArgumentException("audience cannot be null or empty"));
        }

        CompletableFuture<Void> pipelineResult = new CompletableFuture<>();

        try (TurnContextImpl context = new TurnContextImpl(this, reference.getContinuationActivity())) {
            context.getTurnState().add(BOT_IDENTITY_KEY, claimsIdentity);
            context.getTurnState().add(OAUTH_SCOPE_KEY, audience);

            return createConnectorClient(reference.getServiceUrl(), claimsIdentity, audience)
                .thenCompose(connectorClient -> {
                    context.getTurnState().add(CONNECTOR_CLIENT_KEY, connectorClient);
                    return runPipeline(context, callback);
                });
        } catch (Exception e) {
            pipelineResult.completeExceptionally(e);
        }

        return pipelineResult;
    }

    /**
     * Adds middleware to the adapter's pipeline.
     *
     * Middleware is added to the adapter at initialization time. For each turn, the
     * adapter calls middleware in the order in which you added it.
     *
     * @param middleware The middleware to add.
     * @return The updated adapter object.
     */
    public BotFrameworkAdapter use(Middleware middleware) {
        getMiddlewareSet().use(middleware);
        return this;
    }

    /**
     * Creates a turn context and runs the middleware pipeline for an incoming
     * activity.
     *
     * @param authHeader The HTTP authentication header of the request.
     * @param activity   The incoming activity.
     * @param callback   The code to run at the end of the adapter's middleware
     *                   pipeline.
     * @return A task that represents the work queued to execute. If the activity
     *         type was 'Invoke' and the corresponding key (channelId + activityId)
     *         was found then an InvokeResponse is returned, otherwise null is
     *         returned.
     * @throws IllegalArgumentException Activity is null.
     */
    public CompletableFuture<InvokeResponse> processActivity(
        String authHeader,
        Activity activity,
        BotCallbackHandler callback
    ) {
        if (activity == null) {
            return Async.completeExceptionally(new IllegalArgumentException("Activity"));
        }

        return JwtTokenValidation
            .authenticateRequest(activity, authHeader, credentialProvider, channelProvider, authConfiguration)
            .thenCompose(claimsIdentity -> processActivity(claimsIdentity, activity, callback));
    }

    /**
     * Creates a turn context and runs the middleware pipeline for an incoming
     * activity.
     *
     * @param identity A {@link ClaimsIdentity} for the request.
     * @param activity The incoming activity.
     * @param callback The code to run at the end of the adapter's middleware
     *                 pipeline.
     * @return A task that represents the work queued to execute. If the activity
     *         type was 'Invoke' and the corresponding key (channelId + activityId)
     *         was found then an InvokeResponse is returned, otherwise null is
     *         returned.
     * @throws IllegalArgumentException Activity is null.
     */
    public CompletableFuture<InvokeResponse> processActivity(
        ClaimsIdentity identity,
        Activity activity,
        BotCallbackHandler callback
    ) {
        if (activity == null) {
            return Async.completeExceptionally(new IllegalArgumentException("Activity"));
        }

        CompletableFuture<InvokeResponse> pipelineResult = new CompletableFuture<>();

        try (TurnContextImpl context = new TurnContextImpl(this, activity)) {
            activity.setCallerId(generateCallerId(identity).join());
            context.getTurnState().add(BOT_IDENTITY_KEY, identity);

            // The OAuthScope is also stored on the TurnState to get the correct
            // AppCredentials if fetching a token is required.
            String scope = SkillValidation.isSkillClaim(identity.claims())
                ? String.format("%s/.default", JwtTokenValidation.getAppIdFromClaims(identity.claims()))
                : getBotFrameworkOAuthScope();

            context.getTurnState().add(OAUTH_SCOPE_KEY, scope);

            pipelineResult = createConnectorClient(activity.getServiceUrl(), identity, scope)

                // run pipeline
                .thenCompose(connectorClient -> {
                    context.getTurnState().add(CONNECTOR_CLIENT_KEY, connectorClient);
                    return runPipeline(context, callback);
                })
                .thenCompose(result -> {
                    // Handle ExpectedReplies scenarios where the all the activities have been
                    // buffered and sent back at once in an invoke response.
                    if (
                        DeliveryModes
                            .fromString(context.getActivity().getDeliveryMode()) == DeliveryModes.EXPECT_REPLIES
                    ) {
                        return CompletableFuture.completedFuture(
                            new InvokeResponse(
                                HttpURLConnection.HTTP_OK,
                                new ExpectedReplies(context.getBufferedReplyActivities())
                            )
                        );
                    }

                    // Handle Invoke scenarios, which deviate from the request/response model in
                    // that the Bot will return a specific body and return code.
                    if (activity.isType(ActivityTypes.INVOKE)) {
                        Activity invokeResponse = context.getTurnState().get(INVOKE_RESPONSE_KEY);
                        if (invokeResponse == null) {
                            return CompletableFuture
                                .completedFuture(new InvokeResponse(HttpURLConnection.HTTP_NOT_IMPLEMENTED, null));
                        } else {
                            return CompletableFuture.completedFuture((InvokeResponse) invokeResponse.getValue());
                        }
                    }

                    // For all non-invoke scenarios, the HTTP layers above don't have to mess
                    // with the Body and return codes.
                    return CompletableFuture.completedFuture(null);
                });
        } catch (Exception e) {
            pipelineResult.completeExceptionally(e);
        }

        return pipelineResult;
    }

    @SuppressWarnings("PMD")
    private CompletableFuture<String> generateCallerId(ClaimsIdentity claimsIdentity) {
        return credentialProvider.isAuthenticationDisabled().thenApply(is_auth_disabled -> {
            // Is the bot accepting all incoming messages?
            if (is_auth_disabled) {
                return null;
            }

            // Is the activity from another bot?
            if (SkillValidation.isSkillClaim(claimsIdentity.claims())) {
                return String.format(
                    "%s%s",
                    CallerIdConstants.BOT_TO_BOT_PREFIX,
                    JwtTokenValidation.getAppIdFromClaims(claimsIdentity.claims())
                );
            }

            // Is the activity from Public Azure?
            if (channelProvider == null || channelProvider.isPublicAzure()) {
                return CallerIdConstants.PUBLIC_AZURE_CHANNEL;
            }

            // Is the activity from Azure Gov?
            if (channelProvider != null && channelProvider.isGovernment()) {
                return CallerIdConstants.US_GOV_CHANNEL;
            }

            // Return null so that the callerId is cleared.
            return null;
        });
    }

    /**
     * Sends activities to the conversation.
     *
     * @param context    The context object for the turn.
     * @param activities The activities to send.
     * @return A task that represents the work queued to execute. If the activities
     *         are successfully sent, the task result contains an array of
     *         {@link ResourceResponse} objects containing the IDs that the
     *         receiving channel assigned to the activities.
     *
     *         {@link TurnContext#onSendActivities(SendActivitiesHandler)}
     */
    @SuppressWarnings("checkstyle:EmptyBlock, checkstyle:linelength")
    @Override
    public CompletableFuture<ResourceResponse[]> sendActivities(TurnContext context, List<Activity> activities) {
        if (context == null) {
            return Async.completeExceptionally(new IllegalArgumentException("context"));
        }

        if (activities == null) {
            return Async.completeExceptionally(new IllegalArgumentException("activities"));
        }

        if (activities.size() == 0) {
            return Async.completeExceptionally(
                new IllegalArgumentException("Expecting one or more activities, but the array was empty.")
            );
        }

        return CompletableFuture.supplyAsync(() -> {
            ResourceResponse[] responses = new ResourceResponse[activities.size()];

            /*
             * NOTE: we're using for here (vs. foreach) because we want to simultaneously
             * index into the activities array to get the activity to process as well as use
             * that index to assign the response to the responses array and this is the most
             * cost effective way to do that.
             */
            for (int index = 0; index < activities.size(); index++) {
                Activity activity = activities.get(index);

                // Clients and bots SHOULD NOT include an id field in activities they generate.
                activity.setId(null);

                ResourceResponse response;
                if (activity.isType(ActivityTypes.DELAY)) {
                    // The Activity Schema doesn't have a delay type build in, so it's simulated
                    // here in the Bot. This matches the behavior in the Node connector.
                    int delayMs = (int) activity.getValue();
                    try {
                        Thread.sleep(delayMs);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    // No need to create a response. One will be created below.
                    response = null;
                } else if (activity.isType(ActivityTypes.INVOKE_RESPONSE)) {
                    context.getTurnState().add(INVOKE_RESPONSE_KEY, activity);
                    // No need to create a response. One will be created below.
                    response = null;
                } else if (
                    activity.isType(ActivityTypes.TRACE)
                        && !StringUtils.equals(activity.getChannelId(), Channels.EMULATOR)
                ) {
                    // if it is a Trace activity we only send to the channel if it's the emulator.
                    response = null;
                } else if (!StringUtils.isEmpty(activity.getReplyToId())) {
                    ConnectorClient connectorClient = context.getTurnState().get(CONNECTOR_CLIENT_KEY);
                    response = connectorClient.getConversations().replyToActivity(activity).join();
                } else {
                    ConnectorClient connectorClient = context.getTurnState().get(CONNECTOR_CLIENT_KEY);
                    response = connectorClient.getConversations().sendToConversation(activity).join();
                }

                // If No response is set, then default to a "simple" response. This can't really
                // be done above, as there are cases where the ReplyTo/SendTo methods will also
                // return null (See below) so the check has to happen here.
                //
                // Note: In addition to the Invoke / Delay / Activity cases, this code also
                // applies with Skype and Teams with regards to typing events. When sending a
                // typing event in these channels they do not return a RequestResponse which
                // causes the bot to blow up.
                //
                // https://github.com/Microsoft/botbuilder-dotnet/issues/460
                // bug report : https://github.com/Microsoft/botbuilder-dotnet/issues/465
                if (response == null) {
                    response = new ResourceResponse((activity.getId() == null) ? "" : activity.getId());
                }

                responses[index] = response;
            }

            return responses;
        }, ExecutorFactory.getExecutor());
    }

    /**
     * Replaces an existing activity in the conversation.
     *
     * @param context  The context object for the turn.
     * @param activity New replacement activity.
     * @return A task that represents the work queued to execute. If the activity is
     *         successfully sent, the task result contains a
     *         {@link ResourceResponse} object containing the ID that the receiving
     *         channel assigned to the activity.
     *         <p>
     *         Before calling this, set the ID of the replacement activity to the ID
     *         of the activity to replace.
     *         </p>
     *         {@link TurnContext#onUpdateActivity(UpdateActivityHandler)}
     */
    @Override
    public CompletableFuture<ResourceResponse> updateActivity(TurnContext context, Activity activity) {
        ConnectorClient connectorClient = context.getTurnState().get(CONNECTOR_CLIENT_KEY);
        return connectorClient.getConversations().updateActivity(activity);
    }

    /**
     * Deletes an existing activity in the conversation.
     *
     * @param context   The context object for the turn.
     * @param reference Conversation reference for the activity to delete.
     * @return A task that represents the work queued to execute.
     *         {@link TurnContext#onDeleteActivity(DeleteActivityHandler)}
     */
    @Override
    public CompletableFuture<Void> deleteActivity(TurnContext context, ConversationReference reference) {
        ConnectorClient connectorClient = context.getTurnState().get(CONNECTOR_CLIENT_KEY);
        return connectorClient.getConversations()
            .deleteActivity(reference.getConversation().getId(), reference.getActivityId());
    }

    /**
     * Deletes a member from the current conversation.
     *
     * @param context  The context object for the turn.
     * @param memberId ID of the member to delete from the conversation
     * @return A task that represents the work queued to execute.
     */
    public CompletableFuture<Void> deleteConversationMember(TurnContextImpl context, String memberId) {
        if (context.getActivity().getConversation() == null) {
            return Async.completeExceptionally(
                new IllegalArgumentException("BotFrameworkAdapter.deleteConversationMember(): missing conversation")
            );
        }

        if (StringUtils.isEmpty(context.getActivity().getConversation().getId())) {
            return Async.completeExceptionally(
                new IllegalArgumentException("BotFrameworkAdapter.deleteConversationMember(): missing conversation.id")
            );
        }

        ConnectorClient connectorClient = context.getTurnState().get(CONNECTOR_CLIENT_KEY);
        String conversationId = context.getActivity().getConversation().getId();
        return connectorClient.getConversations().deleteConversationMember(conversationId, memberId);
    }

    /**
     * Lists the members of a given activity.
     *
     * @param context The context object for the turn.
     * @return List of Members of the activity
     */
    public CompletableFuture<List<ChannelAccount>> getActivityMembers(TurnContextImpl context) {
        return getActivityMembers(context, null);
    }

    /**
     * Lists the members of a given activity.
     *
     * @param context    The context object for the turn.
     * @param activityId (Optional) Activity ID to enumerate. If not specified the
     *                   current activities ID will be used.
     * @return List of Members of the activity
     */
    public CompletableFuture<List<ChannelAccount>> getActivityMembers(TurnContextImpl context, String activityId) {
        // If no activity was passed in, use the current activity.
        if (activityId == null) {
            activityId = context.getActivity().getId();
        }

        if (context.getActivity().getConversation() == null) {
            return Async.completeExceptionally(
                new IllegalArgumentException("BotFrameworkAdapter.GetActivityMembers(): missing conversation")
            );
        }

        if (StringUtils.isEmpty(context.getActivity().getConversation().getId())) {
            return Async.completeExceptionally(
                new IllegalArgumentException("BotFrameworkAdapter.GetActivityMembers(): missing conversation.id")
            );
        }

        ConnectorClient connectorClient = context.getTurnState().get(CONNECTOR_CLIENT_KEY);
        String conversationId = context.getActivity().getConversation().getId();

        return connectorClient.getConversations().getActivityMembers(conversationId, activityId);
    }

    /**
     * Lists the members of the current conversation.
     *
     * @param context The context object for the turn.
     * @return List of Members of the current conversation
     */
    public CompletableFuture<List<ChannelAccount>> getConversationMembers(TurnContextImpl context) {
        if (context.getActivity().getConversation() == null) {
            return Async.completeExceptionally(
                new IllegalArgumentException("BotFrameworkAdapter.GetActivityMembers(): missing conversation")
            );
        }

        if (StringUtils.isEmpty(context.getActivity().getConversation().getId())) {
            return Async.completeExceptionally(
