/*
 * 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.camel.component.grpc;

import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;

import io.grpc.BindableService;
import io.grpc.Server;
import io.grpc.ServerInterceptor;
import io.grpc.ServerInterceptors;
import io.grpc.netty.GrpcSslContexts;
import io.grpc.netty.NegotiationType;
import io.grpc.netty.NettyServerBuilder;
import io.netty.handler.ssl.ClientAuth;
import io.netty.handler.ssl.SslContextBuilder;
import org.apache.camel.AsyncCallback;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.component.grpc.auth.jwt.JwtServerInterceptor;
import org.apache.camel.component.grpc.server.BindableServiceFactory;
import org.apache.camel.component.grpc.server.DefaultBindableServiceFactory;
import org.apache.camel.component.grpc.server.GrpcHeaderInterceptor;
import org.apache.camel.support.CamelContextHelper;
import org.apache.camel.support.DefaultConsumer;
import org.apache.camel.support.ResourceHelper;
import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static org.apache.camel.component.grpc.GrpcConstants.GRPC_BINDABLE_SERVICE_FACTORY_NAME;

/**
 * Represents gRPC server consumer implementation
 */
public class GrpcConsumer extends DefaultConsumer {

    private static final Logger LOG = LoggerFactory.getLogger(GrpcConsumer.class);

    protected final GrpcConfiguration configuration;
    protected final GrpcEndpoint endpoint;

    private Server server;
    private BindableServiceFactory factory;

    public GrpcConsumer(GrpcEndpoint endpoint, Processor processor, GrpcConfiguration configuration) {
        super(endpoint, processor);
        this.endpoint = endpoint;
        this.configuration = configuration;
    }

    public GrpcConfiguration getConfiguration() {
        return configuration;
    }

    /**
     * Returns the actual local port the gRPC server is listening on. Useful when the server was started with port 0
     * (OS-assigned port).
     */
    public int getLocalPort() {
        return server != null ? server.getPort() : -1;
    }

    @Override
    protected void doStart() throws Exception {
        super.doStart();
        if (server == null) {
            LOG.info("Starting the gRPC server");
            initializeServer();
            server.start();
            LOG.info("gRPC server started and listening on port: {}", server.getPort());
        }
    }

    @Override
    protected void doStop() throws Exception {
        if (server != null) {
            LOG.debug("Terminating gRPC server");
            server.shutdown().shutdownNow();
            server = null;
        }
        super.doStop();
    }

    protected void initializeServer() throws Exception {
        NettyServerBuilder serverBuilder;
        BindableService bindableService = getBindableServiceFactory().createBindableService(this);
        ServerInterceptor headerInterceptor = new GrpcHeaderInterceptor();

        if (ObjectHelper.isNotEmpty(configuration.getHost()) && configuration.getPort() >= 0) {
            LOG.debug("Building gRPC server on {}:{}", configuration.getHost(), configuration.getPort());
            serverBuilder
                    = NettyServerBuilder.forAddress(new InetSocketAddress(configuration.getHost(), configuration.getPort()));
        } else {
            throw new IllegalArgumentException("No server start properties (host, port) specified");
        }

        if (configuration.isRouteControlledStreamObserver()
                && configuration.getConsumerStrategy() == GrpcConsumerStrategy.AGGREGATION) {
            throw new IllegalArgumentException(
                    "Consumer strategy AGGREGATION and routeControlledStreamObserver are not compatible. Set the consumer strategy to PROPAGATION or DELEGATION");
        }

        if (configuration.getNegotiationType() == NegotiationType.TLS) {
            ObjectHelper.notNull(configuration.getKeyCertChainResource(), "keyCertChainResource");
            ObjectHelper.notNull(configuration.getKeyResource(), "keyResource");

            SslContextBuilder sslContextBuilder
                    = SslContextBuilder
                            .forServer(
                                    ResourceHelper.resolveResourceAsInputStream(endpoint.getCamelContext(),
                                            configuration.getKeyCertChainResource()),
                                    ResourceHelper.resolveResourceAsInputStream(endpoint.getCamelContext(),
                                            configuration.getKeyResource()),
                                    configuration.getKeyPassword())
                            .clientAuth(ClientAuth.REQUIRE);

            if (ObjectHelper.isNotEmpty(configuration.getTrustCertCollectionResource())) {
                sslContextBuilder
                        = sslContextBuilder.trustManager(ResourceHelper.resolveResourceAsInputStream(endpoint.getCamelContext(),
                                configuration.getTrustCertCollectionResource()));
            }

            serverBuilder = serverBuilder.sslContext(GrpcSslContexts.configure(sslContextBuilder).build());
        }

        if (configuration.getAuthenticationType() == GrpcAuthType.JWT) {
            ObjectHelper.notNull(configuration.getJwtSecret(), "jwtSecret");

            serverBuilder = serverBuilder.intercept(new JwtServerInterceptor(
                    configuration.getJwtAlgorithm(), configuration.getJwtSecret(),
                    configuration.getJwtIssuer(), configuration.getJwtSubject()));
        }

        // To configure RST_STREAM settings we need valid configuration for both maxRstFramesPerWindow & maxRstPeriodSeconds
        if (configuration.getMaxRstFramesPerWindow() > 0 && configuration.getMaxRstPeriodSeconds() <= 0) {
            throw new IllegalArgumentException("maxRstPeriodSeconds must be a positive value");
        } else if (configuration.getMaxRstFramesPerWindow() <= 0 && configuration.getMaxRstPeriodSeconds() > 0) {
            throw new IllegalArgumentException("maxRstFramesPerWindow must be a positive value");
        } else if (configuration.getMaxRstFramesPerWindow() > 0 && configuration.getMaxRstPeriodSeconds() > 0) {
            serverBuilder.maxRstFramesPerWindow(configuration.getMaxRstFramesPerWindow(),
                    configuration.getMaxRstPeriodSeconds());
        }

        for (ServerInterceptor si : configuration.getServerInterceptors()) {
            serverBuilder.intercept(si);
        }

        server = serverBuilder.addService(ServerInterceptors.intercept(bindableService, headerInterceptor))
                .maxInboundMessageSize(configuration.getMaxMessageSize())
                .flowControlWindow(configuration.getFlowControlWindow())
                .maxConcurrentCallsPerConnection(configuration.getMaxConcurrentCallsPerConnection())
                .initialFlowControlWindow(configuration.getInitialFlowControlWindow())
                .keepAliveTime(configuration.getKeepAliveTime(), TimeUnit.MILLISECONDS)
                .keepAliveTimeout(configuration.getKeepAliveTimeout(), TimeUnit.MILLISECONDS)
                .maxConnectionAge(configuration.getMaxConnectionAge(), TimeUnit.MILLISECONDS)
                .maxConnectionIdle(configuration.getMaxConnectionIdle(), TimeUnit.MILLISECONDS)
                .maxConnectionAgeGrace(configuration.getMaxConnectionAgeGrace(), TimeUnit.MILLISECONDS)
                .maxInboundMetadataSize(configuration.getMaxInboundMetadataSize())
                .permitKeepAliveTime(configuration.getPermitKeepAliveTime(), TimeUnit.MILLISECONDS)
                .permitKeepAliveWithoutCalls(configuration.isPermitKeepAliveWithoutCalls())
                .build();
    }

    public boolean process(Exchange exchange, AsyncCallback callback) {
        exchange.getIn().setHeader(GrpcConstants.GRPC_EVENT_TYPE_HEADER, GrpcConstants.GRPC_EVENT_TYPE_ON_NEXT);
        return doSend(exchange, callback);
    }

    public void onCompleted(Exchange exchange) {
        if (configuration.isForwardOnCompleted()) {
            exchange.getIn().setHeader(GrpcConstants.GRPC_EVENT_TYPE_HEADER, GrpcConstants.GRPC_EVENT_TYPE_ON_COMPLETED);
            doSend(exchange, done -> {
            });
        }
    }

    public void onError(Exchange exchange, Throwable error) {
        if (configuration.isForwardOnError()) {
            exchange.getIn().setHeader(GrpcConstants.GRPC_EVENT_TYPE_HEADER, GrpcConstants.GRPC_EVENT_TYPE_ON_ERROR);
            exchange.getIn().setBody(error);

            doSend(exchange, done -> {
            });
        }
    }

    private boolean doSend(Exchange exchange, AsyncCallback callback) {
        if (this.isRunAllowed()) {
            this.getAsyncProcessor().process(exchange, doneSync -> {
                if (exchange.getException() != null) {
                    getExceptionHandler().handleException("Error processing exchange", exchange, exchange.getException());
                }
                callback.done(doneSync);
            });
            return false;
        } else {
            LOG.warn("Consumer not ready to process exchanges. The exchange {} will be discarded", exchange);
            callback.done(true);
            return true;
        }
    }

    private BindableServiceFactory getBindableServiceFactory() {
        CamelContext context = endpoint.getCamelContext();
        if (this.factory == null) {
            // Try to resolve from the registry
            BindableServiceFactory bindableServiceFactory
                    = CamelContextHelper.lookup(context, GRPC_BINDABLE_SERVICE_FACTORY_NAME, BindableServiceFactory.class);
            if (bindableServiceFactory != null) {
                this.factory = bindableServiceFactory;
            } else {
                // Fallback to the default implementation if an alternative is not available
                this.factory = new DefaultBindableServiceFactory();
            }
        }
        return this.factory;
    }
}
