/* Copyright (c) 2024 Huawei Technologies Co., Ltd.
 * openUBMC is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *          http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 */
#include <lua.hpp>
#include <iomanip>
#include <vector>
#include <memory>
#include <stdlib.h>
#include <mutex>
#include <opentelemetry/trace/provider.h>
#include <opentelemetry/sdk/trace/simple_processor.h>
#include <opentelemetry/sdk/trace/simple_processor_factory.h>
#include <opentelemetry/sdk/trace/tracer_provider_factory.h>
#include <opentelemetry/sdk/trace/batch_span_processor_factory.h>
#include <opentelemetry/sdk/trace/batch_span_processor_options.h>
#include "opentelemetry/exporters/otlp/otlp_grpc_exporter_factory.h"
#include <opentelemetry/sdk/resource/resource.h>

#include <cmath>
#include <random>
#include <array>

#include <opentelemetry/sdk/trace/sampler.h>
#include <opentelemetry/sdk/trace/samplers/parent.h>
#include <opentelemetry/sdk/trace/samplers/always_on.h>
#include <opentelemetry/sdk/trace/samplers/always_off.h>
#include <opentelemetry/sdk/trace/samplers/trace_id_ratio.h>
#include <opentelemetry/trace/trace_id.h>
#include <opentelemetry/nostd/span.h>

#if defined(BUILD_TYPE) && defined(BUILD_TYPE_DT) && BUILD_TYPE == BUILD_TYPE_DT
#include <opentelemetry/exporters/ostream/span_exporter_factory.h>
#include <opentelemetry/exporters/ostream/span_exporter.h>
#include <fstream>
#endif
#include "utils/utils.h"


// 配置参数,用于记录和对比参数
struct TracerConfig {
    int max_queue_size = 2048;
    int schedule_delay_millis = 5000;
    int max_export_batch_size = 512;
    
    bool operator==(const TracerConfig& other) const {
        return max_queue_size == other.max_queue_size &&
            schedule_delay_millis == other.schedule_delay_millis &&
            max_export_batch_size == other.max_export_batch_size;
    }
    
    bool operator!=(const TracerConfig& other) const {
        return !(*this == other);
    }
};

namespace Otel {

namespace common = opentelemetry::common;
namespace sdk_trace = opentelemetry::sdk::trace;
namespace trace = opentelemetry::trace;
using namespace opentelemetry::trace;
using opentelemetry::nostd::string_view;
using Link = std::vector<std::pair<SpanContext, std::map<nostd::string_view, common::AttributeValue>>>;

static constexpr size_t MAX_CACHE_SIZE = 15;
static TracerConfig currentConfig;
static std::mutex initMutex;
static std::atomic<bool> hasProvider{false};
static std::unordered_map<std::string, std::chrono::steady_clock::time_point> force_sample_map_cache;

class CustomSampler : public sdk_trace::Sampler {
public:
    CustomSampler() {
        parent_based_sampler = std::make_shared<sdk_trace::ParentBasedSampler>(
            std::make_shared<sdk_trace::AlwaysOffSampler>()
        );
    }

    sdk_trace::SamplingResult ShouldSample(
        const trace::SpanContext &parent_context,
        trace::TraceId trace_id,
        nostd::string_view name,
        trace::SpanKind span_kind,
        const common::KeyValueIterable &attributes,
        const trace::SpanContextKeyValueIterable &links) noexcept override
    {
        double dynamic_ratio = 0.0;    // 强制采样，负载采样率
        double specified_ratio = 0.0;  // 对象采样率，全局采样率
        parse_sampling_params(attributes, dynamic_ratio, specified_ratio);

        auto result = parent_based_sampler->ShouldSample(
            parent_context, trace_id, name, span_kind, attributes, links);
        
        if (result.decision == sdk_trace::Decision::DROP) {
            // dynamic_ratio = 1.0为强制创建场景
            if (dynamic_ratio == 1.0) {
                return get_always_on_sampler()->ShouldSample(
                    parent_context, trace_id, name, span_kind, attributes, links);
            }
            
            // 对于 Server/Consumer 类型，使用 ratio sampler
            if (span_kind == SpanKind::kServer || span_kind == SpanKind::kConsumer) {
                result = get_ratio_sampler((specified_ratio > 0.0) ? specified_ratio : dynamic_ratio)->ShouldSample(
                    parent_context, trace_id, name, span_kind, attributes, links);
            }
        }
        return result;
    }

    nostd::string_view GetDescription() const noexcept override {
        return "CustomSampler(ParentBased & Ratio{0.000000} for Server/Consumer)";
    }
private:
    void parse_sampling_params(const common::KeyValueIterable &attributes, double &dynamic_ratio, double &specified_ratio) {
        // 只解析，不删除
        attributes.ForEachKeyValue([&](nostd::string_view k, const common::AttributeValue &v) {
            if (k == "otel.dynamic.ratio") {
                dynamic_ratio = extract_double_value(v);
            } else if (k == "otel.specified.ratio") {
                specified_ratio = extract_double_value(v);
            }
            return true;
        });
    }

    static double extract_double_value(const common::AttributeValue &v) {
        if (nostd::holds_alternative<double>(v)) {
            return nostd::get<double>(v);
        } else if (nostd::holds_alternative<int64_t>(v)) {
            return static_cast<double>(nostd::get<int64_t>(v));
        } else if (nostd::holds_alternative<int>(v)) {
            return static_cast<double>(nostd::get<int>(v));
        }
        return 0.0;
    }

    std::shared_ptr<sdk_trace::AlwaysOnSampler> get_always_on_sampler() {
        static std::shared_ptr<sdk_trace::AlwaysOnSampler> always_on_sampler = std::make_shared<sdk_trace::AlwaysOnSampler>();
        return always_on_sampler;
    }

    std::shared_ptr<sdk_trace::TraceIdRatioBasedSampler> get_ratio_sampler(double ratio) {
        std::lock_guard<std::mutex> lock(ratio_cache_mutex);
        double normalized_ratio = std::round(ratio * 1000000.0) / 1000000.0;
        auto it = ratio_sampler_cache.find(normalized_ratio);
        if (it != ratio_sampler_cache.end()) {
            return it->second;
        } else {
            auto sampler = std::make_shared<sdk_trace::TraceIdRatioBasedSampler>(normalized_ratio);
            if (ratio_sampler_cache.size() >= MAX_CACHE_SIZE) {
                ratio_sampler_cache.erase(ratio_sampler_cache.begin());
            }
            ratio_sampler_cache[normalized_ratio] = sampler;
            return sampler;
        }
    }

    std::shared_ptr<sdk_trace::Sampler> parent_based_sampler;
    std::unordered_map<double, std::shared_ptr<sdk_trace::TraceIdRatioBasedSampler>> ratio_sampler_cache;
    std::mutex ratio_cache_mutex;
};

// 获取全局采样器实例（线程安全，仅初始化一次）
static std::shared_ptr<CustomSampler> get_global_sampler() {
    static std::once_flag init_flag;
    static std::shared_ptr<CustomSampler> sampler;
    
    std::call_once(init_flag, [&]() {
        sampler = std::make_shared<CustomSampler>();
    });
    
    return sampler;
}

struct LuaTracer {
    nostd::shared_ptr<trace::Tracer> tracer;
    ~LuaTracer() {
    }
};

struct LuaSpan {
    nostd::shared_ptr<trace::Span> span;
    ~LuaSpan() {
    }
};

namespace LTracer {

static std::string ToHex(const uint8_t* data, size_t size) {
    static const char hex[] = "0123456789abcdef";
    std::string output;
    output.reserve(size * 2);
    for (size_t i = 0; i < size; ++i) {
        output.push_back(hex[data[i] >> 4]);
        output.push_back(hex[data[i] & 0x0F]);
    }
    return output;
}

static bool isHexChar(char c) {
    return (c >= '0' && c <= '9') || (c >= 'a' && c <= 'f') || (c >= 'A' && c <= 'F');
}

static uint8_t toHexValue(char c) {
    if (c >= '0' && c <= '9') {
        return c - '0';
    } else if (c >= 'a' && c <= 'f') {
        return c - 'a' + 10;
    } else if (c >= 'A' && c <= 'F') {
        return c - 'A' + 10;
    } else {
        return 0;
    }
}

static std::vector<uint8_t> HexToBytes(const std::string& hex) {
    std::vector<uint8_t> bytes;
    size_t len = hex.length();
    if (len % 2 != 0) {
        // 长度非偶
        return {};
    }
    bytes.reserve(len / 2);
    for (size_t i = 0; i < len; i += 2) {
        char ch1 = hex[i];
        char ch2 = hex[i + 1];

        if (!isHexChar(ch1) || !isHexChar(ch2)) {
            // 非法字符
            return {};
        }
        uint8_t high = toHexValue(ch1);
        uint8_t low = toHexValue(ch2);
        uint8_t byte = (high << 4) | low;

        bytes.push_back(byte);
    }
    return bytes;
}

static int push_context(lua_State* L, const SpanContext& ctx)
{
    lua_newtable(L); // 创建一个新表

    // trace_id
    auto trace_id = ctx.trace_id();
    std::string trace_id_str = ToHex(trace_id.Id().data(), trace_id.Id().size());
    lua_pushstring(L, "trace_id");
    lua_pushstring(L, trace_id_str.c_str());
    lua_settable(L, -3);

    // span_id
    auto span_id = ctx.span_id();
    std::string span_id_str = ToHex(span_id.Id().data(), span_id.Id().size());
    lua_pushstring(L, "span_id");
    lua_pushstring(L, span_id_str.c_str());
    lua_settable(L, -3);

    // trace_flags
    lua_pushstring(L, "trace_flags");
    lua_pushinteger(L, static_cast<int>(ctx.trace_flags().flags()));
    lua_settable(L, -3);

    // is_remote
    lua_pushstring(L, "is_remote");
    lua_pushboolean(L, ctx.IsRemote());
    lua_settable(L, -3);

    // trace_state
    std::string trace_state_str = ctx.trace_state()->ToHeader();
    lua_pushstring(L, "trace_state");
    lua_pushstring(L, trace_state_str.c_str());
    lua_settable(L, -3);

    return 1;
}

static trace::TraceId GetTraceId(lua_State* L) {
    const char* trace_id_str = lua_tostring(L, -1);
    if (trace_id_str == nullptr) {
        luaL_error(L, "%s: invalid trace id", __FUNCTION__);
    }
    lua_pop(L, 1);
    auto trace_id_bytes = HexToBytes(trace_id_str);
    if (trace_id_bytes.empty() || trace_id_bytes.size() != 16) {
        luaL_error(L, "%s: invalid trace id", __FUNCTION__);
    }
    return trace::TraceId(trace_id_bytes);
}

static trace::SpanId GetSpanId(lua_State* L, int index) {
    lua_getfield(L, index, "span_id");
    const char* span_id_str = lua_tostring(L, -1);
    if (span_id_str == nullptr) {
        luaL_error(L, "%s: invalid span id", __FUNCTION__);
    }
    lua_pop(L, 1);
    auto span_id_bytes = HexToBytes(span_id_str);
    if (span_id_bytes.empty() || span_id_bytes.size() != 8) {
        luaL_error(L, "%s: invalid span id", __FUNCTION__);
    }

    return trace::SpanId(span_id_bytes);
}

static uint8_t GetTraceFlags(lua_State* L, int index) {
    lua_getfield(L, index, "trace_flags");
    uint8_t trace_flags = 0;
    if (lua_isinteger(L, -1)) {
        trace_flags = static_cast<uint8_t>(lua_tointeger(L, -1));
    } else {
        luaL_error(L, "%s: invalid trace flags", __FUNCTION__);
    }
    lua_pop(L, 1);
    return trace_flags;
}

static bool GetIsRemote(lua_State* L, int index) {
    lua_getfield(L, index, "is_remote");
    bool is_remote = lua_isboolean(L, -1) ? lua_toboolean(L, -1) : false;
    lua_pop(L, 1);
    return is_remote;
}

static nostd::shared_ptr<TraceState> GetTraceState(lua_State* L, int index) {
    lua_getfield(L, index, "trace_state");
    const char* trace_state_str = lua_tostring(L, -1);
    if (trace_state_str == nullptr) {
        luaL_error(L, "%s: invalid trace state", __FUNCTION__);
    }
    lua_pop(L, 1);
    auto trace_state = nostd::shared_ptr<TraceState>(trace::TraceState::FromHeader(trace_state_str));
    if (!trace_state) {
        luaL_error(L, "%s: invalid trace state", __FUNCTION__);
    }

    return trace_state;
}

static SpanContext restore_context(lua_State* L, int index) {
    if (index < 0) {
        index = lua_gettop(L) + index + 1;
    }

    lua_getfield(L, index, "trace_id");
    if (lua_isnil(L, -1)) {
        lua_pop(L, 1);
        return SpanContext::GetInvalid();
    }
    TraceId trace_id = GetTraceId(L);
    SpanId span_id = GetSpanId(L, index);
    uint8_t trace_flags = GetTraceFlags(L, index);
    bool is_remote = GetIsRemote(L, index);
    nostd::shared_ptr<TraceState> trace_state = GetTraceState(L, index);

    trace::SpanContext ctx(
        trace_id,
        span_id,
        trace::TraceFlags(trace_flags),
        is_remote,
        trace_state
    );
    return ctx;
}

#if defined(BUILD_TYPE) && defined(BUILD_TYPE_DT) && BUILD_TYPE == BUILD_TYPE_DT
// 创建输出流导出器
std::string file_path = "./span_test.txt";
static std::ofstream file(file_path);
std::ostream& sout = file;

static void init_tracer_provider(int max_size=2048, int delay_millies=5000, int export_size=512) 
{
    std::lock_guard<std::mutex> guard(initMutex);
    if (hasProvider.load(std::memory_order_acquire)) {
        return;
    }
    Utils::initialize_telemetry();
    auto exporter = opentelemetry::exporter::trace::OStreamSpanExporterFactory::Create(sout);

    // 创建输出流导出器
    auto processor = std::unique_ptr<sdk_trace::SpanProcessor>(
        new sdk_trace::SimpleSpanProcessor(std::move(exporter)));

    // 创建TracerProvider
    auto provider = sdk_trace::TracerProviderFactory::Create(std::move(processor));

    // 设置为全局Provider
    trace::Provider::SetTracerProvider(std::move(provider));
    hasProvider.store(true, std::memory_order_release);
}

static int flush(lua_State *L)
{
    sout.flush();
    return 0;
}

#else

static void init_tracer_provider(int max_size=2048, int delay_millies=5000, int export_size=512) 
{
    std::lock_guard<std::mutex> guard(initMutex);
    
    // 创建新的配置
    TracerConfig newConfig{max_size, delay_millies, export_size};
    
    // 如果配置没有变化且已经初始化过，则直接返回
    if (hasProvider.load(std::memory_order_acquire) && currentConfig == newConfig) {
        return;
    }
    
    Utils::initialize_telemetry();
    // 创建grpc导出器
    opentelemetry::exporter::otlp::OtlpGrpcExporterOptions opts;
    opts.endpoint = "localhost:44318";
    auto exporter = opentelemetry::exporter::otlp::OtlpGrpcExporterFactory::Create(opts);

    sdk_trace::BatchSpanProcessorOptions options{};
    options.max_queue_size = max_size;
    options.schedule_delay_millis = std::chrono::milliseconds(delay_millies);
    options.max_export_batch_size = export_size;
    
    // 创建处理器
    auto processor = sdk_trace::BatchSpanProcessorFactory::Create(std::move(exporter), options);

    // 创建TracerProvider
    auto provider = sdk_trace::TracerProviderFactory::Create(std::move(processor));

    // 设置为全局Provider
    trace::Provider::SetTracerProvider(std::move(provider));
    
    // 更新当前配置
    currentConfig = newConfig;
    hasProvider.store(true, std::memory_order_release);
}


#endif

static int l_init_tracer_provider(lua_State *L)
{
    // 最大缓冲span数量,未指定使用默认值2048
    int max_queue_size = lua_isinteger(L, 1) ? static_cast<int32_t>(lua_tointeger(L, 1)) : 2048;

    // 发送间隔,未指定使用默认值5000ms
    int schedule_delay_millis = lua_isinteger(L, 2) ? static_cast<int32_t>(lua_tointeger(L, 2)) : 5000;

    // 设置每次发送最大Span数量,未指定使用默认值512
    int max_export_batch_size = lua_isinteger(L, 3) ? static_cast<int32_t>(lua_tointeger(L, 3)) : 512;

    // 配置TracerProvider
    init_tracer_provider(max_queue_size, schedule_delay_millis, max_export_batch_size);

    return 0;
}

static int l_deinit_tracer_provider(lua_State *L) 
{
    std::lock_guard<std::mutex> guard(initMutex);
    
    if (!hasProvider.load(std::memory_order_acquire)) {
        return 0;
    }
    hasProvider.store(false, std::memory_order_release);
    std::shared_ptr<trace::TracerProvider> none;
    trace::Provider::SetTracerProvider(none);
    currentConfig = TracerConfig{};
    return 0;
}

static int get_tracer(lua_State *L)
{
    if (!hasProvider.load(std::memory_order_acquire)) {
        init_tracer_provider();
    }
    const char *name = lua_tostring(L, 1);
    const char *version = lua_tostring(L, 2);
    const char *url = lua_tostring(L, 3);
    if (name == nullptr || version == nullptr || url == nullptr) {
        luaL_error(L, "%s: invalid parameter, get tracer failed", __FUNCTION__);
    }

    auto provider = trace::Provider::GetTracerProvider();
    auto tracer = provider->GetTracer(name, version, url);
    void* mem = lua_newuserdata(L, sizeof(LuaTracer));
    LuaTracer* udata = new (mem) LuaTracer();
    udata->tracer = tracer;
    luaL_getmetatable(L, "Tracer");
    lua_setmetatable(L, -2);

    return 1;
}

static int lua_tracer_gc(lua_State *L) {
    LuaTracer* lua_tracer = (LuaTracer*)luaL_checkudata(L, 1, "Tracer");
    if (lua_tracer) {
        lua_tracer->~LuaTracer();
    }
    return 0;
}

// 解析 parent 字段
static SpanContext parse_parent(lua_State *L, StartSpanOptions &options) {
    lua_getfield(L, 4, "parent");
    SpanContext parent_ctx = SpanContext::GetInvalid();
    if (!lua_isnil(L, -1)) {
        parent_ctx = restore_context(L, -1);
        if (parent_ctx.IsValid()) {
            options.parent = parent_ctx;
        }
    }
    lua_pop(L, 1);
    return parent_ctx;
}

// 解析 kind 字段
static void parse_kind(lua_State *L, StartSpanOptions &options) {
    // 静态哈希表，用于快速查找span kind字符串对应的枚举值
    static const std::unordered_map<std::string, SpanKind> kind_map = {
        {"Server", SpanKind::kServer},
        {"Client", SpanKind::kClient},
        {"Producer", SpanKind::kProducer},
        {"Consumer", SpanKind::kConsumer},
        {"Internal", SpanKind::kInternal}
    };
    
    lua_getfield(L, 4, "kind");
    if (!lua_isnil(L, -1)) {
        if (!lua_isstring(L, -1)) {
            luaL_error(L, "%s: kind must be a string", __FUNCTION__);
        }
        
        const char* kind_str = lua_tostring(L, -1);
        auto it = kind_map.find(kind_str);
        if (it != kind_map.end()) {
            options.kind = it->second;
        } else {
            luaL_error(L, "%s: invalid span kind value: %s", __FUNCTION__, kind_str);
        }
    } else {
        options.kind = SpanKind::kInternal;
    }
    lua_pop(L, 1);
}

// 解析 force_sample 字段
static bool parse_force_sample(lua_State *L, AttributeMap &attributes, const char* span_name) {
    bool force_sample = false;
    lua_getfield(L, 4, "force_sample");
    if (!lua_isnil(L, -1)) {
        if (!lua_isboolean(L, -1)) {
            luaL_error(L, "%s: force_sample must be boolean", __FUNCTION__);
        }
        
        bool force_sample = lua_toboolean(L, -1);
        if (force_sample) {
            // 在这里进行缓存机制判断
            auto now = std::chrono::steady_clock::now();
            std::string span_name_str(span_name);
            auto it = force_sample_map_cache.find(span_name_str);
            if (it == force_sample_map_cache.end() ||
                std::chrono::duration_cast<std::chrono::seconds>(now - it->second).count() > 300) {
                force_sample_map_cache[span_name_str] = now;
                // 当force_sample为true且通过缓存检查时，设置span.ratio为1
                attributes["otel.dynamic.ratio"] = 1.0;
                force_sample = true;
            }
        }
    }
    lua_pop(L, 1);
    return force_sample;
}

// 解析 ratio 字段
static void parse_ratio(lua_State *L, AttributeMap &attributes) {
    lua_getfield(L, 4, "dynamic_ratio");
    if (!lua_isnil(L, -1)) {
        if (!lua_isnumber(L, -1)) {
            luaL_error(L, "%s: dynamic_ratio must be number", __FUNCTION__);
        }
        
        double dynamic_ratio = lua_tonumber(L, -1);
        attributes["otel.dynamic.ratio"] = dynamic_ratio;
    }
    lua_pop(L, 1);

    lua_getfield(L, 4, "specified_ratio");
    if (!lua_isnil(L, -1)) {
        if (!lua_isnumber(L, -1)) {
            luaL_error(L, "%s: specified_ratio must be number", __FUNCTION__);
        }
        double specified_ratio = lua_tonumber(L, -1);
        attributes["otel.specified.ratio"] = specified_ratio;
    }
    lua_pop(L, 1);
}

// options格式 {parent = parent_ctx, kind = 'Server', ratio = 0.5, force_sample = true}
static SpanContext parse_options(lua_State *L, StartSpanOptions &options, AttributeMap &attributes, const char* span_name) 
{
    if (!lua_istable(L, 4)) {
        options.kind = SpanKind::kInternal;
        return SpanContext::GetInvalid();
    }
    
    SpanContext parent_ctx = parse_parent(L, options);
    parse_kind(L, options);
    bool force_sample = parse_force_sample(L, attributes, span_name);
    if (!force_sample) {
        parse_ratio(L, attributes);
    }
    return parent_ctx;
}

static opentelemetry::trace::TraceId GenerateRandomTraceId() {
    std::array<uint8_t, 16> bytes;
    std::random_device rd;
    std::mt19937 gen(rd());
    std::uniform_int_distribution<uint16_t> dis(0, 255);
    for (auto &b : bytes) {
        b = static_cast<uint8_t>(dis(gen));
    }

    opentelemetry::nostd::span<const uint8_t, 16> span(bytes.data(), 16);
    return opentelemetry::trace::TraceId(span);
}

static bool check_span_sample(lua_State *L, const char *name, SpanContext parent_ctx, StartSpanOptions &options, AttributeMap &attributes) {
    auto sampler = get_global_sampler();
    opentelemetry::trace::TraceId trace_id;
    if (parent_ctx.IsValid()) {
        trace_id = parent_ctx.trace_id();
    } else {
        trace_id = GenerateRandomTraceId();
    }
    std::vector<std::pair<SpanContext, AttributeMap>> empty_links;
    auto result = sampler->ShouldSample(
        parent_ctx,
        trace_id,
        nostd::string_view(name),
        options.kind,
        opentelemetry::common::KeyValueIterableView<AttributeMap>(attributes),
        opentelemetry::trace::SpanContextKeyValueIterableView<std::vector<std::pair<SpanContext, AttributeMap>>>(empty_links)
    );
    return result.decision != sdk_trace::Decision::DROP;
}

static int start_span(lua_State *L)
{
    LuaTracer* lua_tracer = (LuaTracer*)luaL_checkudata(L, 1, "Tracer");
    const char *name = lua_tostring(L, 2);
    if (name == nullptr) {
        luaL_error(L, "%s: invalid parameter, start span failed", __FUNCTION__);
    }

    AttributeMap attributes;
    Utils::get_attributes(L, attributes, 3);

    trace::StartSpanOptions options;
    SpanContext parent_ctx = parse_options(L, options, attributes, name);

    if (!check_span_sample(L, name, parent_ctx, options, attributes)) {
        lua_pushnil(L);
        return 1;
    }
    auto span = lua_tracer->tracer->StartSpan(name, attributes, options);
    void* mem = lua_newuserdata(L, sizeof(LuaSpan));
    LuaSpan* udata = new (mem) LuaSpan();
    udata->span = span;
    luaL_getmetatable(L, "Span");
    lua_setmetatable(L, -2);

    return 1;
}

} // namespace LTracer

namespace LSpan {

static int set_attribute(lua_State *L)
{
    LuaSpan* lua_span = (LuaSpan*)luaL_checkudata(L, 1, "Span");

    const char *key = lua_tostring(L, 2);
    if (key == nullptr) {
        luaL_error(L, "%s: invalid parameter, set attribute failed", __FUNCTION__);
    }

    AttributeValue value = Utils::get_attribute_value(L, 3);
    lua_span->span->SetAttribute(key, value);
    return 0;
}

static int add_event(lua_State *L)
{
    LuaSpan* lua_span = (LuaSpan*)luaL_checkudata(L, 1, "Span");

    const char *name = lua_tostring(L, 2);
    if (name == nullptr) {
        luaL_error(L, "%s: invalid parameter, add event failed", __FUNCTION__);
    }

    AttributeMap attributes;
    if (lua_gettop(L) >= 3) {
        Utils::get_attributes(L, attributes, 3);
    }

    lua_span->span->AddEvent(name, attributes);
    return 0;
}

static int add_link(lua_State *L)
{
    LuaSpan* lua_span = (LuaSpan*)luaL_checkudata(L, 1, "Span");

    SpanContext ctx = LTracer::restore_context(L, 2);

    AttributeMap attributes;
    if (lua_gettop(L) >= 3) {
        Utils::get_attributes(L, attributes, 3);
    }

    lua_span->span->AddLink(ctx, attributes);
    return 0;
}

static const std::unordered_map<std::string, StatusCode> StatusCodeMap = {
    {"unset", StatusCode::kUnset},
    {"ok", StatusCode::kOk},
    {"error", StatusCode::kError}
};

static int set_status(lua_State *L)
{
    LuaSpan* lua_span = (LuaSpan*)luaL_checkudata(L, 1, "Span");

    StatusCode code = StatusCode::kUnset;
    const char *status = lua_tostring(L, 2);
    const char *desc = lua_tostring(L, 3);
    if (status == nullptr || desc == nullptr) {
        luaL_error(L, "%s: invalid parameter, set status failed", __FUNCTION__);
    }

    auto target = StatusCodeMap.find(status);
    if (target == StatusCodeMap.end()) {
        luaL_error(L, "%s: invalid parameter, set status failed", __FUNCTION__);
    } else {
        code = target->second;
    }

    lua_span->span->SetStatus(code, desc);
    return 0;
}

static int get_context(lua_State *L)
{
    LuaSpan* lua_span = (LuaSpan*)luaL_checkudata(L, 1, "Span");

    SpanContext spanctx = lua_span->span->GetContext();
    return LTracer::push_context(L, spanctx);
}

static int is_recording(lua_State *L)
{
    LuaSpan* lua_span = (LuaSpan*)luaL_checkudata(L, 1, "Span");

    if (lua_span->span->IsRecording()) {
        lua_pushboolean(L, 1);
    } else {
        lua_pushboolean(L, 0);
    }

    return 1;
}

static int end(lua_State *L)
{
    LuaSpan* lua_span = (LuaSpan*)luaL_checkudata(L, 1, "Span");

    lua_span->span->End();
    return 0;
}

static int lua_span_gc(lua_State *L) {
    LuaSpan* lua_span = (LuaSpan*)luaL_checkudata(L, 1, "Span");
    if (lua_span) {
        lua_span->~LuaSpan();
    }
    return 0;
}

} // namespace LSpan

#if defined(BUILD_TYPE) && defined(BUILD_TYPE_DT) && BUILD_TYPE == BUILD_TYPE_DT
static const luaL_Reg lmetry[] = {
    {"get_tracer", LTracer::get_tracer},
    {"flush", LTracer::flush},
    {"init", LTracer::l_init_tracer_provider},
    {"deinit", LTracer::l_deinit_tracer_provider},
    {NULL, NULL}
};

#else

static const luaL_Reg lmetry[] = {
    {"get_tracer", LTracer::get_tracer},
    {"init", LTracer::l_init_tracer_provider},
    {"deinit", LTracer::l_deinit_tracer_provider},
    {NULL, NULL}
};

#endif

static const luaL_Reg ltracer[] = {
    {"start_span", LTracer::start_span},
    {NULL, NULL}
};

static const luaL_Reg lspan[] = {
    {"set_attribute", LSpan::set_attribute},
    {"add_event", LSpan::add_event},
    {"add_link", LSpan::add_link},
    {"set_status", LSpan::set_status},
    {"get_context", LSpan::get_context},
    {"is_recording", LSpan::is_recording},
    {"finish", LSpan::end},
    {NULL, NULL}
};

static void set_index_metatable(lua_State *L)
{
    luaL_newmetatable(L, "Tracer");
    luaL_setfuncs(L, ltracer, 0);
    lua_pushvalue(L, -1);
    lua_setfield(L, -2, "__index");

    lua_pushcfunction(L, LTracer::lua_tracer_gc);
    lua_setfield(L, -2, "__gc");

    luaL_newmetatable(L, "Span");
    luaL_setfuncs(L, lspan, 0);
    lua_pushvalue(L, -1);
    lua_setfield(L, -2, "__index");

    lua_pushcfunction(L, LSpan::lua_span_gc);
    lua_setfield(L, -2, "__gc");
}

extern "C" int luaopen_otel_trace(lua_State *L)
{
    luaL_checkversion(L);
    set_index_metatable(L);
    luaL_newlibtable(L, lmetry);
    luaL_setfuncs(L, lmetry, 0);

    return 1;
}

} // namespace Otel