package com.atlassian.bitbucket.mesh.grpc;

import com.atlassian.bitbucket.mesh.rpc.v1.*;
import com.atlassian.bitbucket.mesh.throttle.ResourceBusyException;
import com.atlassian.bitbucket.mesh.throttle.ThrottleService;
import com.atlassian.bitbucket.mesh.throttle.ThrottledResource;
import com.atlassian.bitbucket.mesh.throttle.Ticket;
import com.google.common.collect.ImmutableMap;
import com.google.protobuf.DescriptorProtos.MethodOptions;
import com.google.protobuf.DescriptorProtos.ServiceOptions;
import com.google.protobuf.Descriptors;
import io.grpc.*;
import io.grpc.protobuf.ProtoMethodDescriptorSupplier;

import java.util.Map;

/**
 * Inspects the {@link MethodOptions} and {@link ServiceOptions} for the gRPC request and
 * {@link ThrottleService#acquireTicket acquires a ticket} if an {@link RpcThrottledResource}
 * is specified in the method or service options.
 */
public class ThrottlingServerInterceptor implements ServerInterceptor {

    private static final Map<RpcThrottledResource, ThrottledResource> RESOURCE_MAPPING = ImmutableMap.of(
            RpcThrottledResource.RESOURCE_COMMAND, ThrottledResource.COMMAND,
            RpcThrottledResource.RESOURCE_LFS, ThrottledResource.LFS,
            RpcThrottledResource.RESOURCE_MIRROR_PACK, ThrottledResource.MIRROR_PACK,
            //RpcThrottledResource.RESOURCE_NONE, null, //This implicit mapping allows opting out of throttling
            RpcThrottledResource.RESOURCE_PACK, ThrottledResource.PACK,
            RpcThrottledResource.RESOURCE_REFS, ThrottledResource.REFS
    );

    private final ThrottleService throttleService;

    public ThrottlingServerInterceptor(ThrottleService throttleService) {
        this.throttleService = throttleService;
    }

    @Override
    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call, Metadata headers,
                                                                 ServerCallHandler<ReqT, RespT> next) {
        ServerCall.Listener<ReqT> listener = next.startCall(call, headers);

        RpcThrottledResource requestedResource = getThrottledResource(call);
        ThrottledResource throttledResource = RESOURCE_MAPPING.get(requestedResource);
        if (throttledResource == null) {
            return listener;
        }

        //A ServerCall.Listener is used rather than acquiring the ticket here because if we throw an exception
        //from this method the RPC is not cleaned up properly. Waiting for the first message avoids that
        return new ForwardingServerCallListener.SimpleForwardingServerCallListener<ReqT>(listener) {

            private volatile boolean callClosed;
            private volatile Ticket ticket;

            @Override
            public void onHalfClose() {
                try {
                    super.onHalfClose();
                } catch (IllegalStateException e) {
                    if (!callClosed) {
                        //If we've closed the call in onMessage because we failed to acquire a ticket, this
                        //IllegalStateException is expected and does not have to be logged
                        throw e;
                    }
                }
            }

            @Override
            public void onMessage(ReqT message) {
                try {
                    if (ticket == null) {
                        //Acquire the ticket on the first message
                        ticket = throttleService.acquireTicket(throttledResource);
                    }
                    super.onMessage(message);
                } catch (ResourceBusyException e) {
                    if (!callClosed) {
                        callClosed = true;
                        call.close(Status.RESOURCE_EXHAUSTED, RpcErrorUtils.toMetadata(RpcError.newBuilder()
                                .setResourceBusy(RpcResourceBusy.newBuilder()
                                        .setResource(requestedResource)
                                        .build())
                                .build()));
                    }
                }
            }

            @Override
            public void onComplete() {
                try {
                    super.onComplete();
                } finally {
                    if (ticket != null) {
                        ticket.close();
                    }
                }
            }
        };
    }

    private static Descriptors.MethodDescriptor getMethodDescriptor(ServerCall<?, ?> call) {
        Object schemaDescriptor = call.getMethodDescriptor().getSchemaDescriptor();
        if (schemaDescriptor instanceof ProtoMethodDescriptorSupplier) {
            return ((ProtoMethodDescriptorSupplier) schemaDescriptor).getMethodDescriptor();
        }
        return null;
    }

    private static RpcThrottledResource getThrottledResource(ServerCall<?, ?> call) {
        Descriptors.MethodDescriptor methodDescriptor = getMethodDescriptor(call);
        if (methodDescriptor == null) {
            return null;
        }

        MethodOptions methodOptions = methodDescriptor.getOptions();
        if (methodOptions.hasExtension(RpcProtos.methodThrottle)) {
            return methodOptions.getExtension(RpcProtos.methodThrottle);
        }

        ServiceOptions serviceOptions = methodDescriptor.getService().getOptions();
        if (serviceOptions.hasExtension(RpcProtos.serviceThrottle)) {
            return serviceOptions.getExtension(RpcProtos.serviceThrottle);
        }

        return null;
    }
}
