#include <softbus_storage/StorageService.h>

#include <filesystem>

namespace softbus::storage {

namespace fs = std::filesystem;

StorageService::StorageService(std::string dataDir)
    : dataDir_(std::move(dataDir))
{
}

bool StorageService::open()
{
    fs::create_directories(dataDir_);
    timeseries_.open(dataDir_ + "/timeseries.jsonl", std::ios::app);
    commandLog_.open(dataDir_ + "/command_log.jsonl", std::ios::app);
    eventLog_.open(dataDir_ + "/event_log.jsonl", std::ios::app);
    return timeseries_.is_open() && commandLog_.is_open() && eventLog_.is_open();
}

void StorageService::attachBus(softbus::bus::Bus* bus)
{
    bus_ = bus;
    if (!bus_) {
        return;
    }
    open();
    dataSub_ = bus_->subscribe("data/#", [this](const softbus::bus::BusMessage& msg) {
        onBusMessage(msg);
    });
    ackSub_ = bus_->subscribe("control/ack/#", [this](const softbus::bus::BusMessage& msg) {
        onBusMessage(msg);
    });
    eventSub_ = bus_->subscribe("event/#", [this](const softbus::bus::BusMessage& msg) {
        onBusMessage(msg);
    });
}

void StorageService::onBusMessage(const softbus::bus::BusMessage& message)
{
    switch (message.type) {
    case softbus::bus::BusPayloadType::DataPoint:
        appendLine("timeseries", message.dataPoint.toJson().dump());
        break;
    case softbus::bus::BusPayloadType::Ack:
        appendLine("command", message.ack.toJson().dump());
        break;
    case softbus::bus::BusPayloadType::Event:
        appendLine("event", message.event.toJson().dump());
        break;
    default:
        break;
    }
}

void StorageService::appendLine(const std::string& fileName, const std::string& line)
{
    std::lock_guard<std::mutex> lock(mutex_);
    if (fileName == "timeseries" && timeseries_.is_open()) {
        timeseries_ << line << '\n';
    } else if (fileName == "command" && commandLog_.is_open()) {
        commandLog_ << line << '\n';
    } else if (fileName == "event" && eventLog_.is_open()) {
        eventLog_ << line << '\n';
    }
}

} // namespace softbus::storage
