#include <softbus_pipeline/PipelineEngine.h>

#include <softbus_core/logging/Logger.h>
#include <softbus_core/time/HwClock.h>

#include <vector>

namespace softbus::pipeline {

PipelineEngine::PipelineEngine(softbus::core::PluginLoader* plugin,
                               softbus::opcua::UaModelBinder* binder,
                               softbus::registry::PointRegistry* registry)
    : plugin_(plugin)
    , binder_(binder)
    , registry_(registry)
{
}

void PipelineEngine::attachBus(softbus::bus::Bus* bus)
{
    bus_ = bus;
    if (!bus_) {
        return;
    }
    rawSub_ = bus_->subscribe("edge/raw/#", [this](const softbus::bus::BusMessage& msg) {
        onBusMessage(msg);
    });
}

void PipelineEngine::onBusMessage(const softbus::bus::BusMessage& message)
{
    if (message.type == softbus::bus::BusPayloadType::RawPacket) {
        process(message.rawPacket);
    }
}

void PipelineEngine::process(const softbus::core::RawPacket& packet)
{
    if (!plugin_ || !plugin_->parser() || !binder_) {
        return;
    }

    const uint8_t* data = packet.payload.data();
    std::size_t size = packet.payload.size();
    std::vector<uint8_t> frame = packet.payload;

    if (plugin_->framer()) {
        const auto framed = plugin_->framer()->extract(data, size);
        if (framed.success) {
            frame = framed.frame;
            data = frame.data();
            size = frame.size();
        }
    }

    auto parsed = plugin_->parser()->parseFrame(data, size, packet.sourceId);
    if (!parsed.success) {
        SB_LOG_WARN_TRACE("PipelineEngine", packet.traceId, "Failed to parse RawPacket");
        return;
    }

    parsed.message.traceId = packet.traceId;
    parsed.message.timestampNs = packet.timestampNs;

    if (registry_) {
        const auto entry = registry_->findByObjectRef(parsed.message.id);
        if (entry) {
            publishDataPoint(parsed.message, *entry);
        }
    }

    binder_->bind(parsed.message);
    SB_LOG_INFO_TRACE("PipelineEngine", packet.traceId,
        "RawPacket converted to DOMMessage id=", parsed.message.id);
}

void PipelineEngine::publishDataPoint(const softbus::core::DOMMessage& message,
                                      const softbus::registry::PointTableEntry& entry)
{
    if (!bus_) {
        return;
    }

    double value = 0.0;
    if (message.value.contains("value")) {
        value = message.value["value"].get<double>();
    } else if (message.value.contains("registerValue")) {
        value = message.value["registerValue"].get<double>();
    }

    softbus::core::SoftbusDataPoint dp;
    dp.deviceId = entry.deviceId;
    dp.pointId = entry.pointId;
    dp.value = value;
    dp.unit = entry.unit;
    dp.objectRef = entry.objectRef;
    dp.sourceTimestampNs = message.timestampNs;

    softbus::bus::BusMessage busMsg;
    busMsg.type = softbus::bus::BusPayloadType::DataPoint;
    busMsg.dataPoint = dp;
    const std::string topic = entry.dataTopic.empty()
        ? registry_->topicRules().formatData(entry.deviceId, entry.pointId)
        : entry.dataTopic;
    bus_->publish(topic, busMsg);
}

} // namespace softbus::pipeline
