/*
 * 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.shardingsphere.proxy.frontend.netty;

import com.google.common.hash.Hashing;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.authority.model.ShardingSpherePrivileges;
import org.apache.shardingsphere.authority.rule.AuthorityRule;
import org.apache.shardingsphere.data.pipeline.cdc.context.CDCConnectionContext;
import org.apache.shardingsphere.data.pipeline.cdc.exception.CDCExceptionWrapper;
import org.apache.shardingsphere.data.pipeline.cdc.exception.CDCLoginFailedException;
import org.apache.shardingsphere.data.pipeline.cdc.exception.EmptyCDCLoginRequestBodyException;
import org.apache.shardingsphere.data.pipeline.cdc.generator.CDCResponseUtils;
import org.apache.shardingsphere.data.pipeline.cdc.handler.CDCBackendHandler;
import org.apache.shardingsphere.data.pipeline.cdc.protocol.request.AckStreamingRequestBody;
import org.apache.shardingsphere.data.pipeline.cdc.protocol.request.CDCRequest;
import org.apache.shardingsphere.data.pipeline.cdc.protocol.request.DropStreamingRequestBody;
import org.apache.shardingsphere.data.pipeline.cdc.protocol.request.LoginRequestBody.BasicBody;
import org.apache.shardingsphere.data.pipeline.cdc.protocol.request.StartStreamingRequestBody;
import org.apache.shardingsphere.data.pipeline.cdc.protocol.request.StopStreamingRequestBody;
import org.apache.shardingsphere.data.pipeline.cdc.protocol.request.StreamDataRequestBody;
import org.apache.shardingsphere.data.pipeline.cdc.protocol.response.CDCResponse;
import org.apache.shardingsphere.data.pipeline.cdc.protocol.response.CDCResponse.Status;
import org.apache.shardingsphere.data.pipeline.cdc.protocol.response.ServerGreetingResult;
import org.apache.shardingsphere.data.pipeline.core.exception.param.PipelineInvalidParameterException;
import org.apache.shardingsphere.infra.autogen.version.ShardingSphereVersion;
import org.apache.shardingsphere.infra.exception.core.ShardingSpherePreconditions;
import org.apache.shardingsphere.infra.exception.core.external.sql.sqlstate.XOpenSQLState;
import org.apache.shardingsphere.infra.exception.core.external.sql.type.kernel.category.PipelineSQLException;
import org.apache.shardingsphere.infra.exception.dialect.SQLExceptionTransformEngine;
import org.apache.shardingsphere.infra.exception.dialect.exception.syntax.database.UnknownDatabaseException;
import org.apache.shardingsphere.infra.exception.kernel.metadata.rule.MissingRequiredRuleException;
import org.apache.shardingsphere.infra.exception.mysql.exception.AccessDeniedException;
import org.apache.shardingsphere.infra.metadata.user.Grantee;
import org.apache.shardingsphere.infra.metadata.user.ShardingSphereUser;
import org.apache.shardingsphere.proxy.backend.context.ProxyContext;
import org.apache.shardingsphere.proxy.frontend.protocol.FrontDatabaseProtocolTypeFactory;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.sql.SQLException;
import java.util.Objects;

/**
 * CDC channel inbound handler.
 */
@Slf4j
public final class CDCChannelInboundHandler extends ChannelInboundHandlerAdapter {
    
    private static final AttributeKey<CDCConnectionContext> CONNECTION_CONTEXT_KEY = AttributeKey.valueOf("connection.context");
    
    private final CDCBackendHandler backendHandler = new CDCBackendHandler();
    
    @Override
    public void channelActive(final ChannelHandlerContext ctx) {
        CDCResponse response = CDCResponse.newBuilder().setServerGreetingResult(ServerGreetingResult.newBuilder().setServerVersion(ShardingSphereVersion.VERSION).setProtocolVersion("1").build())
                .setStatus(Status.SUCCEED).build();
        ctx.writeAndFlush(response);
    }
    
    @Override
    public void channelInactive(final ChannelHandlerContext ctx) {
        CDCConnectionContext connectionContext = ctx.channel().attr(CONNECTION_CONTEXT_KEY).get();
        if (null != connectionContext && null != connectionContext.getJobId()) {
            backendHandler.stopStreaming(connectionContext.getJobId(), ctx.channel().id());
        }
        ctx.channel().attr(CONNECTION_CONTEXT_KEY).set(null);
    }
    
    @Override
    public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cause) {
        log.error("caught CDC resolution error", cause);
        // TODO add CDC exception to wrapper this exception, and add the parameters requestId and whether to close connect
        ChannelFuture channelFuture;
        if (cause instanceof CDCExceptionWrapper) {
            CDCExceptionWrapper wrapper = (CDCExceptionWrapper) cause;
            SQLException sqlException = SQLExceptionTransformEngine.toSQLException(wrapper.getCause(), FrontDatabaseProtocolTypeFactory.getDatabaseType());
            channelFuture = ctx.writeAndFlush(CDCResponseUtils.failed(wrapper.getRequestId(), sqlException.getSQLState(), sqlException.getMessage()));
        } else {
            channelFuture = ctx.writeAndFlush(CDCResponseUtils.failed("", XOpenSQLState.GENERAL_ERROR.getValue(), String.valueOf(cause.getMessage())));
        }
        CDCConnectionContext connectionContext = ctx.channel().attr(CONNECTION_CONTEXT_KEY).get();
        if (null == connectionContext) {
            channelFuture.addListener(ChannelFutureListener.CLOSE);
        }
    }
    
    @Override
    public void channelRead(final ChannelHandlerContext ctx, final Object msg) {
        CDCConnectionContext connectionContext = ctx.channel().attr(CONNECTION_CONTEXT_KEY).get();
        CDCRequest request = (CDCRequest) msg;
        if (null == connectionContext || request.hasLoginRequestBody()) {
            processLogin(ctx, request);
            return;
        }
        switch (request.getType()) {
            case STREAM_DATA:
                processStreamDataRequest(ctx, request, connectionContext);
                break;
            case ACK_STREAMING:
                processAckStreamingRequest(request);
                break;
            case STOP_STREAMING:
                processStopStreamingRequest(ctx, request, connectionContext);
                break;
            case START_STREAMING:
                processStartStreamingRequest(ctx, request, connectionContext);
                break;
            case DROP_STREAMING:
                processDropStreamingRequest(ctx, request, connectionContext);
                break;
            default:
                log.warn("can't handle this type of request {}", request);
                break;
        }
    }
    
    private void processLogin(final ChannelHandlerContext ctx, final CDCRequest request) {
        ShardingSpherePreconditions.checkState(request.hasLoginRequestBody() && request.getLoginRequestBody().hasBasicBody(),
                () -> new CDCExceptionWrapper(request.getRequestId(), new EmptyCDCLoginRequestBodyException()));
        BasicBody body = request.getLoginRequestBody().getBasicBody();
        AuthorityRule authorityRule = ProxyContext.getInstance().getContextManager().getMetaDataContexts().getMetaData().getGlobalRuleMetaData().getSingleRule(AuthorityRule.class);
        ShardingSphereUser user = authorityRule.findUser(new Grantee(body.getUsername(), getHostAddress(ctx)))
                .orElseThrow(() -> new CDCExceptionWrapper(request.getRequestId(), new CDCLoginFailedException()));
        ShardingSpherePreconditions.checkState(Objects.equals(Hashing.sha256().hashBytes(user.getPassword().getBytes()).toString().toUpperCase(), body.getPassword()),
                () -> new CDCExceptionWrapper(request.getRequestId(), new CDCLoginFailedException()));
        ctx.channel().attr(CONNECTION_CONTEXT_KEY).set(new CDCConnectionContext(user));
        ctx.writeAndFlush(CDCResponseUtils.succeed(request.getRequestId()));
    }
    
    private void checkPrivileges(final String requestId, final Grantee grantee, final String currentDatabase) {
        AuthorityRule authorityRule = ProxyContext.getInstance().getContextManager().getMetaDataContexts().getMetaData().getGlobalRuleMetaData().findSingleRule(AuthorityRule.class)
                .orElseThrow(() -> new CDCExceptionWrapper(requestId, new MissingRequiredRuleException("authority")));
        ShardingSpherePrivileges privileges = authorityRule.findPrivileges(grantee)
                .orElseThrow(() -> new CDCExceptionWrapper(requestId, new AccessDeniedException(grantee.getUsername(), grantee.getHostname(), false)));
        ShardingSpherePreconditions.checkState(privileges.hasPrivileges(currentDatabase),
                () -> new CDCExceptionWrapper(requestId, new UnknownDatabaseException(currentDatabase)));
    }
    
    private String getHostAddress(final ChannelHandlerContext context) {
        SocketAddress socketAddress = context.channel().remoteAddress();
        return socketAddress instanceof InetSocketAddress ? ((InetSocketAddress) socketAddress).getAddress().getHostAddress() : socketAddress.toString();
    }
    
    private void processStreamDataRequest(final ChannelHandlerContext ctx, final CDCRequest request, final CDCConnectionContext connectionContext) {
        if (!request.hasStreamDataRequestBody()) {
            throw new CDCExceptionWrapper(request.getRequestId(), new PipelineInvalidParameterException("Stream data request body is empty"));
        }
        StreamDataRequestBody requestBody = request.getStreamDataRequestBody();
        if (requestBody.getDatabase().isEmpty()) {
            throw new CDCExceptionWrapper(request.getRequestId(), new PipelineInvalidParameterException("Database is empty"));
        }
        if (requestBody.getSourceSchemaTableList().isEmpty()) {
            throw new CDCExceptionWrapper(request.getRequestId(), new PipelineInvalidParameterException("Source schema table is empty"));
        }
        checkPrivileges(request.getRequestId(), connectionContext.getCurrentUser().getGrantee(), requestBody.getDatabase());
        try {
            CDCResponse response = backendHandler.streamData(request.getRequestId(), requestBody, connectionContext, ctx.channel());
            ctx.writeAndFlush(response);
        } catch (final PipelineSQLException ex) {
            throw new CDCExceptionWrapper(request.getRequestId(), ex);
        }
    }
    
    private void processAckStreamingRequest(final CDCRequest request) {
        if (!request.hasAckStreamingRequestBody()) {
            throw new CDCExceptionWrapper(request.getRequestId(), new PipelineInvalidParameterException("Ack request body is empty"));
        }
        AckStreamingRequestBody requestBody = request.getAckStreamingRequestBody();
        if (requestBody.getAckId().isEmpty()) {
            throw new CDCExceptionWrapper(request.getRequestId(), new PipelineInvalidParameterException("Ack request is empty"));
        }
        backendHandler.processAck(requestBody);
    }
    
    private void processStartStreamingRequest(final ChannelHandlerContext ctx, final CDCRequest request, final CDCConnectionContext connectionContext) {
        if (!request.hasStartStreamingRequestBody()) {
            throw new CDCExceptionWrapper(request.getRequestId(), new PipelineInvalidParameterException("Start streaming request body is empty"));
        }
        StartStreamingRequestBody requestBody = request.getStartStreamingRequestBody();
        if (requestBody.getStreamingId().isEmpty()) {
            throw new CDCExceptionWrapper(request.getRequestId(), new PipelineInvalidParameterException("Streaming id is empty"));
        }
        String database = backendHandler.getDatabaseNameByJobId(requestBody.getStreamingId());
        checkPrivileges(request.getRequestId(), connectionContext.getCurrentUser().getGrantee(), database);
        backendHandler.startStreaming(requestBody.getStreamingId(), connectionContext, ctx.channel());
        ctx.writeAndFlush(CDCResponseUtils.succeed(request.getRequestId()));
    }
    
    private void processStopStreamingRequest(final ChannelHandlerContext ctx, final CDCRequest request, final CDCConnectionContext connectionContext) {
        StopStreamingRequestBody requestBody = request.getStopStreamingRequestBody();
        String database = backendHandler.getDatabaseNameByJobId(requestBody.getStreamingId());
        checkPrivileges(request.getRequestId(), connectionContext.getCurrentUser().getGrantee(), database);
        backendHandler.stopStreaming(requestBody.getStreamingId(), ctx.channel().id());
        connectionContext.setJobId(null);
        ctx.writeAndFlush(CDCResponseUtils.succeed(request.getRequestId()));
    }
    
    private void processDropStreamingRequest(final ChannelHandlerContext ctx, final CDCRequest request, final CDCConnectionContext connectionContext) {
        DropStreamingRequestBody requestBody = request.getDropStreamingRequestBody();
        checkPrivileges(request.getRequestId(), connectionContext.getCurrentUser().getGrantee(), backendHandler.getDatabaseNameByJobId(requestBody.getStreamingId()));
        backendHandler.dropStreaming(requestBody.getStreamingId());
        connectionContext.setJobId(null);
        ctx.writeAndFlush(CDCResponseUtils.succeed(request.getRequestId()));
    }
}
