// 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.

#include "runtime/cdc_client_mgr.h"

#include <brpc/closure_guard.h>

#include <fcntl.h>

#include <fmt/core.h>

#include <gen_cpp/internal_service.pb.h>

#include <google/protobuf/stubs/callback.h>

#include <signal.h>

#include <sys/stat.h>

#include <sys/wait.h>

#include <unistd.h>

#include <cstdio>

#ifndef __APPLE__

#include <sys/prctl.h>

#endif

#include <atomic>

#include <chrono>

#include <iterator>

#include <mutex>

#include <sstream>

#include <string>

#include <thread>

#include <vector>

#include "common/config.h"

#include "common/logging.h"

#include "common/status.h"

#include "runtime/exec_env.h"

#include "service/http/http_client.h"

namespace doris {

namespace {

// Handle SIGCHLD signal to prevent zombie processes

void handle_sigchld(int sig_no) {

    int status = 0;

    pid_t pid;

    while ((pid = waitpid(-1, &status, WNOHANG)) > 0) {

    }

}

// Check CDC client health

#ifndef BE_TEST

Status check_cdc_client_health(int retry_times, int sleep_time, std::string& health_response) {

    const std::string cdc_health_url =

            "http://127.0.0.1:" + std::to_string(doris::config::cdc_client_port) +

            "/actuator/health";

    auto health_request = [cdc_health_url, &health_response](HttpClient* client) {

        RETURN_IF_ERROR(client->init(cdc_health_url));

        client->set_timeout_ms(5000);

        RETURN_IF_ERROR(client->execute(&health_response));

        return Status::OK();

    };

    Status status = HttpClient::execute_with_retry(retry_times, sleep_time, health_request);

    if (!status.ok()) {

        return Status::InternalError("CDC client health check failed");

    }

    bool is_up = health_response.find("UP") != std::string::npos;

    if (!is_up) {

        return Status::InternalError(fmt::format("CDC client unhealthy: {}", health_response));

    }

    return Status::OK();

}

#endif

} // anonymous namespace

CdcClientMgr::CdcClientMgr() = default;

CdcClientMgr::~CdcClientMgr() {

    stop();

}

void CdcClientMgr::stop() {

    pid_t pid = _child_pid.load();

    if (pid > 0) {

        // Check if process is still alive

        if (kill(pid, 0) == 0) {

            LOG(INFO) << "Stopping CDC client process, pid=" << pid;

            // Send SIGTERM for graceful shutdown

            kill(pid, SIGTERM);

            // Wait a short time for graceful shutdown

            std::this_thread::sleep_for(std::chrono::milliseconds(200));

            // Force kill if still alive

            if (kill(pid, 0) == 0) {

                LOG(INFO) << "Force killing CDC client process, pid=" << pid;

                kill(pid, SIGKILL);

                int status = 0;

                waitpid(pid, &status, 0);

            }

        }

        _child_pid.store(0);

    }

    LOG(INFO) << "CdcClientMgr is stopped";

}

Status CdcClientMgr::start_cdc_client(PRequestCdcClientResult* result) {

    std::lock_guard<std::mutex> lock(_start_mutex);

    Status st = Status::OK();

    pid_t exist_pid = _child_pid.load();

    if (exist_pid > 0) {

#ifdef BE_TEST

        // In test mode, directly return OK if PID exists

        LOG(INFO) << "cdc client already started (BE_TEST mode), pid=" << exist_pid;

        return Status::OK();

#else

        // Check if process is still alive

        if (kill(exist_pid, 0) == 0) {

            // Process exists, verify it's actually our CDC client by health check

            std::string check_response;

            auto check_st = check_cdc_client_health(3, 1, check_response);

            if (check_st.ok()) {

                // Process exists and responding, CDC client is running

                return Status::OK();

            } else {

                // Process exists but CDC client not responding

                // Either it's a different process (PID reused) or CDC client is unhealthy

                st = Status::InternalError(fmt::format("CDC client {} unresponsive", exist_pid));

                st.to_protobuf(result->mutable_status());

                return st;

            }

        } else {

            LOG(INFO) << "CDC client is dead, pid=" << exist_pid;

            // Process is dead, reset PID and continue to start

            _child_pid.store(0);

        }

#endif

    } else if (!_adopted_external.load()) {

        LOG(INFO) << "CDC client has never been started";

    }

#ifndef BE_TEST

    // Adopt an externally-managed cdc_client if the port already answers

    // healthy (e.g. one started manually for debug / hotfix).

    {

        std::string adopt_response;

        if (check_cdc_client_health(1, 0, adopt_response).ok()) {

            if (!_adopted_external.exchange(true)) {

                LOG(INFO) << "Adopting external cdc client on port "

                          << doris::config::cdc_client_port;

            }

            return Status::OK();

        }

    }

    _adopted_external.store(false);

#endif

    const char* doris_home = getenv("DORIS_HOME");

    const char* log_dir = getenv("LOG_DIR");

    const std::string cdc_jar_path = std::string(doris_home) + "/lib/cdc_client/cdc-client.jar";

    const std::string cdc_jar_port =

            "--server.port=" + std::to_string(doris::config::cdc_client_port);

    const std::string backend_http_port =

            "--backend.http.port=" + std::to_string(config::webserver_port);

    const std::string cluster_token = "--cluster.token=" + ExecEnv::GetInstance()->token();

    const std::string java_opts = "-Dlog.path=" + std::string(log_dir);

    // check cdc jar exists

    struct stat buffer;

    if (stat(cdc_jar_path.c_str(), &buffer) != 0) {

        st = Status::InternalError("Can not find cdc-client.jar.");

        st.to_protobuf(result->mutable_status());

        return st;

    }

    // Ready to start cdc client

    LOG(INFO) << "Ready to start cdc client";

    const auto* java_home = getenv("JAVA_HOME");

    if (!java_home) {

        st = Status::InternalError("Can not find JAVA_HOME");

        st.to_protobuf(result->mutable_status());

        return st;

    }

    std::string path(java_home);

    std::string java_bin = path + "/bin/java";

    // Pre-build everything the child needs before fork(): heap allocation after

    // fork() in a multi-threaded process can deadlock on inherited libc locks.

    std::vector<std::string> argv_storage;

    argv_storage.emplace_back("java");

    const std::string user_java_opts = doris::config::cdc_client_java_opts;

    if (!user_java_opts.empty()) {

        std::istringstream iss(user_java_opts);

        argv_storage.insert(argv_storage.end(), std::istream_iterator<std::string>(iss),

                            std::istream_iterator<std::string>());

    }

    argv_storage.emplace_back(java_opts);

    // OOM safety net (last-wins, user opts cannot disable).

    argv_storage.emplace_back("-XX:+ExitOnOutOfMemoryError");

    // JDK17 opens for debezium ObjectSizeCalculator reflection.

    argv_storage.emplace_back("--add-opens=java.base/java.lang=ALL-UNNAMED");

    argv_storage.emplace_back("--add-opens=java.base/java.util=ALL-UNNAMED");

    argv_storage.emplace_back("--add-opens=java.base/java.math=ALL-UNNAMED");

    argv_storage.emplace_back("--add-opens=java.base/java.nio=ALL-UNNAMED");

    argv_storage.emplace_back("-jar");

    argv_storage.emplace_back(cdc_jar_path);

    argv_storage.emplace_back(cdc_jar_port);

    argv_storage.emplace_back(backend_http_port);

    argv_storage.emplace_back(cluster_token);

    std::vector<char*> argv;

    argv.reserve(argv_storage.size() + 1);

    for (auto& s : argv_storage) {

        argv.push_back(const_cast<char*>(s.c_str()));

    }

    argv.push_back(nullptr);

    const std::string cdc_out_file = std::string(log_dir) + "/cdc-client.out";

    struct sigaction act;

    act.sa_flags = 0;

    act.sa_handler = handle_sigchld;

    sigaction(SIGCHLD, &act, NULL);

    LOG(INFO) << "Start to fork cdc client process with " << path;

#ifdef BE_TEST

    _child_pid.store(99999);

    st = Status::OK();

    return st;

#else

    pid_t pid = fork();

    if (pid < 0) {

        st = Status::InternalError("Fork cdc client failed.");

        st.to_protobuf(result->mutable_status());

        return st;

    } else if (pid == 0) {

        // Child: async-signal-safe operations only until execv().

#ifndef __APPLE__

        prctl(PR_SET_PDEATHSIG, SIGKILL);

#endif

        int out_fd = open(cdc_out_file.c_str(), O_WRONLY | O_CREAT | O_APPEND | O_CLOEXEC, 0644);

        if (out_fd < 0) {

            perror("open cdc-client.out file failed");

            _exit(1);

        }

        dup2(out_fd, STDOUT_FILENO);

        dup2(out_fd, STDERR_FILENO);

        close(out_fd);

        execv(java_bin.c_str(), argv.data());

        perror("Cdc client child process error");

        _exit(1);

    } else {

        // Parent process: save PID and wait for startup

        _child_pid.store(pid);

        // Waiting for cdc to start, failed after more than 3 * 10 seconds

        std::string health_response;

        Status status = check_cdc_client_health(3, 10, health_response);

        if (!status.ok()) {

            // Reset PID if startup failed

            _child_pid.store(0);

            st = Status::InternalError("Start cdc client failed.");

            st.to_protobuf(result->mutable_status());

        } else if (kill(pid, 0) != 0) {

            // Port healthy but our child has exited: an external process is

            // answering. Treat as adoption instead of masking dead PID as success.

            _child_pid.store(0);

            if (!_adopted_external.exchange(true)) {

                LOG(INFO) << "Forked cdc client " << pid << " exited but port "

                          << doris::config::cdc_client_port

                          << " is healthy, adopting external instance";

            }

        } else {

            _adopted_external.store(false);

            LOG(INFO) << "Start cdc client success, pid=" << pid

                      << ", status=" << status.to_string() << ", response=" << health_response;

        }

    }

#endif //BE_TEST

    return st;

}

void CdcClientMgr::request_cdc_client_impl(const PRequestCdcClientRequest* request,

                                           PRequestCdcClientResult* result,

                                           google::protobuf::Closure* done) {

    brpc::ClosureGuard closure_guard(done);

    // Start CDC client if not started

    Status start_st = start_cdc_client(result);

    if (!start_st.ok()) {

        LOG(ERROR) << "Failed to start CDC client, status=" << start_st.to_string();

        start_st.to_protobuf(result->mutable_status());

        return;

    }

    std::string cdc_response;

    Status st = send_request_to_cdc_client(request->api(), request->params(), &cdc_response);

    result->set_response(cdc_response);

    st.to_protobuf(result->mutable_status());

}

Status CdcClientMgr::send_request_to_cdc_client(const std::string& api,

                                                const std::string& params_body,

                                                std::string* response) {

    std::string remote_url_prefix =

            fmt::format("http://127.0.0.1:{}{}", doris::config::cdc_client_port, api);

    auto cdc_request = [&remote_url_prefix, response, &params_body](HttpClient* client) {

        RETURN_IF_ERROR(client->init(remote_url_prefix));

        client->set_timeout_ms(doris::config::request_cdc_client_timeout_ms);

        if (!params_body.empty()) {

            client->set_payload(params_body);

        }

        client->set_content_type("application/json");

        client->set_method(POST);

        RETURN_IF_ERROR(client->execute(response));

        return Status::OK();

    };

    return HttpClient::execute_with_retry(3, 1, cdc_request);

}

} // namespace doris