/*
 * Copyright (C) 2016-2022 ActionTech.
 * License: http://www.gnu.org/licenses/gpl.html GPL version 2 or higher.
 */

package com.actiontech.dble.sqlengine;

import com.actiontech.dble.DbleServer;
import com.actiontech.dble.backend.datasource.PhysicalDbInstance;
import com.actiontech.dble.backend.mysql.nio.handler.ResponseHandler;
import com.actiontech.dble.config.ErrorInfo;
import com.actiontech.dble.net.connection.BackendConnection;
import com.actiontech.dble.net.mysql.ErrorPacket;
import com.actiontech.dble.net.service.AbstractService;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;


/**
 * Created by szf on 2018/9/20.
 */
public class SpecialSqlJob extends SQLJob {

    public static final Logger LOGGER = LoggerFactory.getLogger(SpecialSqlJob.class);

    private final SQLJobHandler jobHandler;
    private final ResponseHandler sqlJob;
    private final PhysicalDbInstance ds;
    private final String schema;
    private final String sql;
    private final List<ErrorInfo> list;
    private final AtomicBoolean finished;

    public SpecialSqlJob(String sql, String schema, SQLJobHandler jobHandler, PhysicalDbInstance ds, List<ErrorInfo> list) {
        super(sql, schema, jobHandler, ds);
        this.jobHandler = jobHandler;
        this.ds = ds;
        this.schema = schema;
        this.sql = sql;
        this.list = list;
        this.sqlJob = this;
        this.finished = new AtomicBoolean(false);
    }


    public void run() {
        //create new connection and exec the sql
        DbleServer.getInstance().getComplexQueryExecutor().execute(new Runnable() {
            @Override
            public void run() {
                try {
                    ds.createConnectionSkipPool(schema, sqlJob);
                } catch (Exception e) {
                    sqlJob.connectionError(e, null);
                }
            }
        });
    }

    @Override
    protected boolean doFinished(boolean failed) {
        if (finished.compareAndSet(false, true)) {
            jobHandler.finished(schema, failed);
            return true;
        }
        return false;
    }


    @Override
    public void connectionError(Throwable e, Object attachment) {
        LOGGER.warn("can't get connection for sql :" + sql, e);
        list.add(new ErrorInfo("Meta", "WARNING", "Can't get connection for Meta check in shardingNode[" + ds.getName() + "." + schema + "]"));
        doFinished(true);
    }

    @Override
    public void connectionAcquired(BackendConnection conn) {
        conn.getBackendService().setResponseHandler(this);
        conn.getBackendService().setComplexQuery(true);
        try {
            conn.getBackendService().query(sql, true);
        } catch (Exception e) { // (UnsupportedEncodingException e) {
            doFinished(true);
        }
    }

    @Override
    public void errorResponse(byte[] err, @NotNull AbstractService service) {
        ErrorPacket errPg = new ErrorPacket();
        errPg.read(err);

        String errMsg = "error response errNo:" + errPg.getErrNo() + ", " + new String(errPg.getMessage()) +
                " from of sql :" + sql + " at con:" + service;

        list.add(new ErrorInfo("Meta", "WARNING", "Execute show tables in shardingNode[" + ds.getName() + "." + schema + "] get error"));

        LOGGER.info(errMsg);
        doFinished(true);
        service.getConnection().close("dry run error backend");
    }

    @Override
    public void connectionClose(@NotNull AbstractService service, String reason) {
        if (doFinished(true)) {
            list.add(new ErrorInfo("Meta", "WARNING", "Execute show tables in shardingNode[" + ds.getName() + "." + schema + "] get connection closed"));
        }
    }


    @Override
    public void rowEofResponse(byte[] eof, boolean isLeft, @NotNull AbstractService service) {
        doFinished(false);
        service.getConnection().close("dry run used connection");
    }

}
