roro9stack/src/services/storage_service.cpp

#include "storage_service.h"

#include <SD.h>
#include <ff.h>
#include <sd_diskio.h>

#include <algorithm>
#include <atomic>
#include <memory>

#include "platform/pins.h"
#include "platform/shared_spi.h"

namespace roro {

void StorageService::start() {
    if (task_) return;
    xTaskCreate(taskEntry, "storage", 6144, this, 1, &task_);  // peak 3.9 KB (SD install with its signature check, M2)
}

void StorageService::stop() {
    if (!task_) return;
    vTaskDelete(task_);
    task_ = nullptr;
    if (mounted_) SD.end();
    mounted_ = false;
}

StorageState StorageService::state() const {
    lock();
    StorageState copy = monitor_.state();
    unlock();
    return copy;
}

void StorageService::appendLine(const std::string& path, const std::string& line, bool capture) {
    lock();
    size_t bytes = path.size() + line.size();
    bool allowed = capture ? monitor_.state().capturesAllowed : monitor_.state().logsAllowed;
    bool accept = allowed && pendingBytes_ + bytes <= kMaxPendingBytes;
    if (accept) {
        pending_.emplace_back(path, line);
        pendingBytes_ += bytes;
    } else {
        dropped_++;
    }
    unlock();
}

void StorageService::requestListing() {
    lock();
    listingRequested_ = true;
    listingReady_ = false;
    unlock();
    if (task_) xTaskNotifyGive(task_);
}

bool StorageService::listingReady() const {
    lock();
    bool ready = listingReady_;
    unlock();
    return ready;
}

std::vector<std::vector<StoredFile>> StorageService::listing() const {
    lock();
    auto copy = listing_;
    unlock();
    return copy;
}

void StorageService::requestDelete(std::vector<std::string> paths) {
    lock();
    toDelete_.insert(toDelete_.end(), paths.begin(), paths.end());
    unlock();
    if (task_) xTaskNotifyGive(task_);
}

void StorageService::runJob(std::function<void()> job) {
    lock();
    jobs_.push_back(std::move(job));
    unlock();
    if (task_) xTaskNotifyGive(task_);
}

bool StorageService::runAndWait(std::function<void()> job) {
    // Shared with the job, which outlives this wait if the card is missing and it never starts.
    // Exactly one side wins the state change: the job (Queued -> Running) or the wait giving up
    // (Queued -> Abandoned), so an abandoned job can never touch this stack frame.
    enum : int { Queued, Running, Abandoned };
    struct Run {
        std::atomic<int> state{Queued};
        SemaphoreHandle_t done = xSemaphoreCreateBinary();
        ~Run() { vSemaphoreDelete(done); }
    };
    auto run = std::make_shared<Run>();
    runJob([run, job]() {
        int expected = Queued;
        if (!run->state.compare_exchange_strong(expected, Running)) return;
        job();
        xSemaphoreGive(run->done);
    });
    for (int waited = 0; xSemaphoreTake(run->done, pdMS_TO_TICKS(500)) != pdTRUE; waited += 500) {
        int expected = Queued;
        if (waited >= 5000 && run->state.compare_exchange_strong(expected, Abandoned)) return false;
    }
    return true;
}

bool StorageService::requestFormat() {
    if (formatRequested_ || !task_) return false;
    formatRequested_ = true;
    xTaskNotifyGive(task_);
    return true;
}

void StorageService::taskEntry(void* self) { static_cast<StorageService*>(self)->loop(); }

void StorageService::loop() {
    uint32_t wakes = kPollEvery;  // poll right away
    for (;;) {
        if (formatRequested_) {
            format();
            formatRequested_ = false;
            wakes = kPollEvery;
        }
        if (wakes++ >= kPollEvery) {
            wakes = 1;
            poll();
        }
        writePending();

        lock();
        bool wantListing = listingRequested_;
        std::vector<std::string> deleting;
        deleting.swap(toDelete_);
        unlock();
        if (!deleting.empty()) {
            remove(deleting);
            poll();
        }
        if (wantListing) list();

        lock();
        std::vector<std::function<void()>> jobs;
        jobs.swap(jobs_);
        unlock();
        for (auto& job : jobs)
            if (mounted_) job();

        // Sleep until the next batch, or until a request wakes us early.
        ulTaskNotifyTake(pdTRUE, pdMS_TO_TICKS(kWakeMs));
    }
}

void StorageService::format() {
    if (mounted_) SD.end();
    mounted_ = false;

    bool ok = false;
    uint8_t pdrv = sdcard_init(pins::kSdCs, &sharedSpi(), 20000000);
    if (pdrv != 0xFF) {
        constexpr size_t kWorkSize = 4096;  // FF_MAX_SS
        std::unique_ptr<uint8_t[]> work(new uint8_t[kWorkSize]);
        LBA_t partitions[] = {100, 0};  // one partition, 100 % of the card
        char drive[3] = {static_cast<char>('0' + pdrv), ':', 0};
        MKFS_PARM options = {FM_FAT32, 0, 0, 0, 0};
        ok = f_fdisk(pdrv, partitions, work.get()) == FR_OK &&
             f_mkfs(drive, &options, work.get(), kWorkSize) == FR_OK;
        sdcard_uninit(pdrv);
    }
    bus_.publish(Event::withText(EventType::Notification, ok ? "SD card formatted" : "SD card format failed",
                                 static_cast<int32_t>(ok ? NotificationLevel::Info : NotificationLevel::Warning)));
}

bool StorageService::mount() {
    return SD.begin(pins::kSdCs, sharedSpi(), 20000000, "/sd", 5, false);
}

void StorageService::poll() {
    if (mounted_) {
        // Detect removal: the root can no longer be opened.
        File root = SD.open("/");
        if (!root) {
            SD.end();
            mounted_ = false;
        }
    }
    if (!mounted_) mounted_ = mount();

    bool present = mounted_;
    uint64_t total = present ? SD.totalBytes() : 0;
    uint64_t used = present ? SD.usedBytes() : 0;

    lock();
    monitor_.update(present, total, used);
    unlock();
}

void StorageService::writePending() {
    lock();
    std::deque<std::pair<std::string, std::string>> batch;
    batch.swap(pending_);
    pendingBytes_ = 0;
    unlock();
    if (batch.empty() || !mounted_) return;

    // Keep each file's lines in order while opening each file once.
    std::stable_sort(batch.begin(), batch.end(), [](const auto& a, const auto& b) { return a.first < b.first; });
    File file;
    std::string openPath;
    for (auto& [path, line] : batch) {
        if (path != openPath) {
            if (file) file.close();
            file = SD.open(path.c_str(), FILE_APPEND, true);  // true: create missing folders
            openPath = path;
        }
        if (file) {
            file.write(reinterpret_cast<const uint8_t*>(line.data()), line.size());
            file.write('\n');
        }
    }
    if (file) file.close();
}

namespace {
void walk(File dir, std::vector<StoredFile>& out) {
    for (File f = dir.openNextFile(); f; f = dir.openNextFile()) {
        if (f.isDirectory()) walk(f, out);
        else out.push_back({f.path(), f.size()});
    }
}
}  // namespace

void StorageService::list() {
    std::vector<std::vector<StoredFile>> result;
    for (auto& category : kCleanupCategories) {
        std::vector<StoredFile> files;
        if (mounted_) {
            File dir = SD.open(category.folder);
            if (dir && dir.isDirectory()) walk(dir, files);
        }
        result.push_back(std::move(files));
    }
    lock();
    listing_ = std::move(result);
    listingRequested_ = false;
    listingReady_ = true;
    unlock();
}

void StorageService::remove(const std::vector<std::string>& paths) {
    uint64_t freed = 0;
    int failed = 0;
    for (auto& p : paths) {
        File f = SD.open(p.c_str());
        uint64_t size = f ? f.size() : 0;
        if (f) f.close();
        if (SD.remove(p.c_str())) freed += size;
        else failed++;
    }
    std::string text = "Freed " + formatBytes(freed) + (failed ? " (" + std::to_string(failed) + " failed)" : "");
    bus_.publish(Event::withText(EventType::Notification, text.c_str(),
                                 static_cast<int32_t>(failed ? NotificationLevel::Warning : NotificationLevel::Info)));
}

}  // namespace roro