package com.atlassian.bitbucket.mesh.grpc;

import com.atlassian.bitbucket.mesh.config.NodeConfig;
import com.atlassian.bitbucket.mesh.rpc.v1.ManagementServiceGrpc;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcError;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcErrorUtils;
import com.atlassian.bitbucket.mesh.rpc.v1.RpcWrongTarget;
import io.grpc.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static com.atlassian.bitbucket.mesh.rpc.MetadataKeys.KEY_REMOTE_CLIENT;
import static com.atlassian.bitbucket.mesh.rpc.MetadataKeys.KEY_TARGET_ID;
import static org.apache.commons.lang3.StringUtils.defaultString;
import static org.apache.commons.lang3.StringUtils.isBlank;

/**
 * Intercepts requests and ensures that this node's ID matches the <em>optional</em> TARGET_ID header specified in
 * the RPC metadata, if such a header is provided. This ensures that the system quickly detects and recovers from
 * situations where a client uses stale DNS data when setting up the RPC connection and happens to use an IP address
 * that was previously assigned to node X, but has now been assigned to node Y.
 * <p>
 * This situation is more likely to occur than one might expect, thanks to two contributing factors:
 * <ul>
 *     <li>grpc-java's clients use DNS cache with a TTL of 30s, which _may_ result in stale DNS data being used
 *     to resolve a hostname.</li>
 *     <li>kubernetes and docker-compose have been observed to shuffle IP addresses around when restarting multiple
 *     instances of a Mesh service.</li>
 * </ul>
 *
 * @since 1.2
 */
public class WrongTargetServerInterceptor implements ServerInterceptor {

    private static final Logger log = LoggerFactory.getLogger(WrongTargetServerInterceptor.class);

    private static final String PROVISION_METHOD = ManagementServiceGrpc.getProvisionNodeMethod().getFullMethodName();

    private final NodeConfig nodeConfig;

    public WrongTargetServerInterceptor(NodeConfig nodeConfig) {
        this.nodeConfig = nodeConfig;
    }

    @Override
    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call, Metadata headers,
                                                                 ServerCallHandler<ReqT, RespT> next) {
        String actualId = nodeConfig.getId();
        String targetId = headers.get(KEY_TARGET_ID);
        if (isBlank(targetId) || targetId.equals(actualId) || actualId.equals(NodeConfig.UNCONFIGURED_NODE_ID) ||
                call.getMethodDescriptor().getFullMethodName().equals(PROVISION_METHOD)) {
            return next.startCall(call, headers);
        }

        //This RPC was intended for another node, reject the request
        log.warn("Rejecting {} call from {} intended for node with ID {} (this node's ID = {}). Has the IP changed?",
                call.getMethodDescriptor().getFullMethodName(), defaultString(headers.get(KEY_REMOTE_CLIENT), "-"),
                targetId, actualId);
        call.close(Status.FAILED_PRECONDITION.withDescription("The provided " + KEY_TARGET_ID.name() +
                        " (" + targetId + ") does not match this node's ID (" + actualId + "). Has the IP changed?"),
                RpcErrorUtils.toMetadata(RpcError.newBuilder()
                        .setWrongTarget(RpcWrongTarget.newBuilder()
                                .setActualId(actualId)
                                .setTargetId(targetId))
                .build()));

        //Nothing to do; we've already rejected the request
        return new ServerCall.Listener<ReqT>() { };
    }
}
