eive-obsw/mission/tmtc/PersistentTmStore.cpp

#include "PersistentTmStore.h"

#include <mission/memory/SdCardMountedIF.h>

#include <algorithm>
#include <cinttypes>
#include <filesystem>
#include <fstream>
#include <utility>

#include "fsfw/ipc/CommandMessage.h"
#include "fsfw/ipc/QueueFactory.h"
#include "fsfw/tmstorage/TmStoreMessage.h"

using namespace returnvalue;

PersistentTmStore::PersistentTmStore(object_id_t objectId, const char* baseDir,
                                     std::string baseName, RolloverInterval intervalUnit,
                                     uint32_t intervalCount, timeval& currentTv,
                                     StorageManagerIF& tmStore, SdCardMountedIF& sdcMan)
    : SystemObject(objectId),
      baseDir(baseDir),
      baseName(std::move(baseName)),
      currentTv(currentTv),
      sdcMan(sdcMan),
      tmStore(tmStore) {
  tcQueue = QueueFactory::instance()->createMessageQueue();
  calcDiffSeconds(intervalUnit, intervalCount);
}

ReturnValue_t PersistentTmStore::handleCommandQueue(StorageManagerIF& ipcStore,
                                                    TmFunnelBase& tmFunnel) {
  CommandMessage cmdMessage;
  ReturnValue_t result = tcQueue->receiveMessage(&cmdMessage);
  if (result == MessageQueueIF::EMPTY) {
    return returnvalue::OK;
  }
  if (result != returnvalue::OK) {
    return result;
  }
  if (cmdMessage.getMessageType() == messagetypes::TM_STORE) {
    Command_t cmd = cmdMessage.getCommand();
    if (cmd == TmStoreMessage::DELETE_STORE_CONTENT_TIME) {
      store_address_t storeId = TmStoreMessage::getStoreId(&cmdMessage);
      auto accessor = ipcStore.getData(storeId);
      uint32_t deleteUpToUnixSeconds = 0;
      size_t size = accessor.second.size();
      SerializeAdapter::deSerialize(&deleteUpToUnixSeconds, accessor.second.data(), &size,
                                    SerializeIF::Endianness::NETWORK);
      deleteUpTo(deleteUpToUnixSeconds);
    } else if (cmd == TmStoreMessage::DOWNLINK_STORE_CONTENT_TIME) {
      store_address_t storeId = TmStoreMessage::getStoreId(&cmdMessage);
      auto accessor = ipcStore.getData(storeId);
      uint32_t dumpFromUnixSeconds;
      uint32_t dumpUntilUnixSeconds;
      size_t size = accessor.second.size();
      SerializeAdapter::deSerialize(&dumpFromUnixSeconds, accessor.second.data(), &size,
                                    SerializeIF::Endianness::NETWORK);
      SerializeAdapter::deSerialize(&dumpUntilUnixSeconds, accessor.second.data(), &size,
                                    SerializeIF::Endianness::NETWORK);
      // TODO: TM store missing, and maybe there is a better way to do this?
      dumpFromUpTo(dumpFromUnixSeconds, dumpUntilUnixSeconds, tmFunnel);
    }
  }
  return returnvalue::OK;
}

ReturnValue_t PersistentTmStore::passPacket(PusTmReader& reader) {
  bool inApidList = false;
  if (filter.apid) {
    auto& apidFilter = filter.apid.value();
    if (std::find(apidFilter.begin(), apidFilter.end(), reader.getApid()) != apidFilter.end()) {
      if (not filter.serviceSubservices and not filter.services) {
        return storePacket(reader);
      }
      inApidList = true;
    }
  }
  std::pair<uint8_t, uint8_t> serviceSubservice;
  serviceSubservice.first = reader.getService();
  serviceSubservice.second = reader.getSubService();
  if (filter.services) {
    auto& serviceFilter = filter.services.value();
    if (std::find(serviceFilter.begin(), serviceFilter.end(), serviceSubservice.first) !=
        serviceFilter.end()) {
      if (filter.apid and inApidList) {
        return storePacket(reader);
      }
    }
  }
  if (filter.serviceSubservices) {
    auto& serviceSubserviceFilter = filter.serviceSubservices.value();
    if (std::find(serviceSubserviceFilter.begin(), serviceSubserviceFilter.end(),
                  serviceSubservice) != serviceSubserviceFilter.end()) {
      if (filter.apid and inApidList) {
        return storePacket(reader);
      }
    }
  }
  return returnvalue::OK;
}

void PersistentTmStore::dumpFrom(uint32_t fromUnixSeconds, TmFunnelBase& tmFunnel) {
  return dumpFromUpTo(fromUnixSeconds, currentTv.tv_sec, tmFunnel);
}

ReturnValue_t PersistentTmStore::storePacket(PusTmReader& reader) {
  using namespace std::filesystem;
  if (baseDirUninitialized) {
    updateBaseDir();
  }
  // It is assumed here that the filesystem is usable.
  if (not mostRecentFile) {
    assignAndOrCreateMostRecentFile();
  }

  if (currentTv.tv_sec < mostRecentTv.value().tv_sec or
      currentTv.tv_sec - mostRecentTv.value().tv_sec > static_cast<int>(rolloverDiffSeconds)) {
    if (file_size(mostRecentFile.value()) + reader.getFullPacketLen() > fileBuf.size()) {
      uint8_t appendedCounter = 1;
      path rolloverName;
      while (true) {
        rolloverName =
            path(mostRecentFile.value().string() + "." + std::to_string(appendedCounter));
        if (not exists(rolloverName)) {
          break;
        }
        appendedCounter++;
      }
      rename(mostRecentFile.value(), rolloverName);
      std::ofstream of(mostRecentFile.value(), std::ios::binary);
    }
  }

  // Rollover conditions were handled, write to file now
  std::ofstream of(mostRecentFile.value(), std::ios::app | std::ios::binary);
  of.write(reinterpret_cast<const char*>(reader.getFullData()),
           static_cast<std::streamsize>(reader.getFullPacketLen()));
  return returnvalue::OK;
}

MessageQueueId_t PersistentTmStore::getCommandQueue() const { return tcQueue->getId(); }

void PersistentTmStore::calcDiffSeconds(RolloverInterval intervalUnit, uint32_t intervalCount) {
  if (intervalUnit == RolloverInterval::MINUTELY) {
    rolloverDiffSeconds = 60 * intervalCount;
  } else if (intervalUnit == RolloverInterval::HOURLY) {
    rolloverDiffSeconds = 60 * 60 * intervalCount;
  } else if (intervalUnit == RolloverInterval::DAILY) {
    rolloverDiffSeconds = 60 * 60 * 24 * intervalCount;
  }
}

void PersistentTmStore::updateBaseDir() {
  using namespace std::filesystem;
  std::string currentPrefix = sdcMan.getCurrentMountPrefix();
  basePath = path(currentPrefix) / baseDir / baseName;
  if (not exists(basePath)) {
    create_directories(basePath);
  }
  baseDirUninitialized = false;
}

void PersistentTmStore::assignAndOrCreateMostRecentFile() {
  using namespace std::filesystem;
  for (auto const& file : directory_iterator(basePath)) {
    if (file.is_directory()) {
      continue;
    }
    auto pathStr = file.path().string();
    Clock::TimeOfDay_t tod;
    if (pathStr.find(baseName) == std::string::npos) {
      continue;
    }
    unsigned int underscorePos = pathStr.find_last_of('_');
    std::string stampStr = pathStr.substr(underscorePos + 1);
    int count = sscanf(stampStr.c_str(),
                       "%04" SCNu32 "-%02" SCNu32 "-%02" SCNu32 "T%02" SCNu32 "-%02" SCNu32
                       "-%02" SCNu32 "Z",
                       &tod.year, &tod.month, &tod.day, &tod.hour, &tod.minute, &tod.second);
    if (count != 6) {
      continue;
    }
    timeval tv{};
    Clock::convertTimeOfDayToTimeval(&tod, &tv);
    if (not mostRecentTv || tv > mostRecentTv.value()) {
      mostRecentTv = tv;
      mostRecentFile = file.path();
    }
  }
  if (not mostRecentFile) {
    unsigned currentIdx = 0;
    path pathStart = basePath / baseName;
    memcpy(fileBuf.data() + currentIdx, pathStart.c_str(), pathStart.string().length());
    currentIdx += pathStart.string().length();
    Clock::TimeOfDay_t tod;
    Clock::convertTimevalToTimeOfDay(&currentTv, &tod);
    currentIdx += sprintf(reinterpret_cast<char*>(fileBuf.data() + currentIdx),
                          "_%04" PRIu32 "-%02" PRIu32 "-%02" PRIu32 "T%02" PRIu32 "-%02" PRIu32
                          "-%02" PRIu32 "Z.bin",
                          tod.year, tod.month, tod.day, tod.hour, tod.minute, tod.second);
    path newPath(std::string(reinterpret_cast<const char*>(fileBuf.data()), currentIdx));
    std::ofstream of(newPath, std::ios::binary);
    mostRecentFile = newPath;
    mostRecentTv = currentTv;
  }
}

void PersistentTmStore::addApid(uint16_t apid) {
  if (not filter.apid) {
    filter.apid = std::vector<uint16_t>({apid});
    return;
  }
  filter.apid.value().push_back(apid);
}

void PersistentTmStore::addService(uint8_t service) {
  if (not filter.services) {
    filter.services = std::vector<uint8_t>({service});
    return;
  }
  filter.services.value().push_back(service);
}

void PersistentTmStore::addServiceSubservice(uint8_t service, uint8_t subservice) {
  if (not filter.serviceSubservices) {
    filter.serviceSubservices =
        std::vector<std::pair<uint8_t, uint8_t>>({std::pair(service, subservice)});
    return;
  }
  filter.serviceSubservices.value().emplace_back(service, subservice);
}

void PersistentTmStore::deleteUpTo(uint32_t unixSeconds) {
  using namespace std::filesystem;
  for (auto const& file : directory_iterator(basePath)) {
    if (file.is_directory() or
        (mostRecentFile.has_value() and (mostRecentFile.value() == file.path()))) {
      continue;
    }
    Clock::TimeOfDay_t tod;
    pathToTod(file.path(), tod);
    timeval time{};
    ReturnValue_t result = Clock::convertTimeOfDayToTimeval(&tod, &time);
    if (result != returnvalue::OK) {
      sif::error << "TOD to time conversion failed for file " << file << std::endl;
      continue;
    }
    if (time.tv_sec + rolloverDiffSeconds < unixSeconds) {
      std::filesystem::remove(file.path());
    }
  }
}

void PersistentTmStore::dumpFromUpTo(uint32_t fromUnixSeconds, uint32_t upToUnixSeconds,
                                     TmFunnelBase& funnel) {
  using namespace std::filesystem;
  for (auto const& file : directory_iterator(basePath)) {
    if (file.is_directory()) {
      continue;
    }
    if (mostRecentFile.has_value() and mostRecentTv.has_value() and
        (file.path() == mostRecentFile.value()) and
        (upToUnixSeconds < static_cast<uint32_t>(mostRecentTv.value().tv_sec))) {
      continue;
    }
    Clock::TimeOfDay_t tod;
    pathToTod(file.path(), tod);
    timeval time{};
    ReturnValue_t result = Clock::convertTimeOfDayToTimeval(&tod, &time);
    if (result != returnvalue::OK) {
      sif::error << "TOD to time conversion failed for file " << file << std::endl;
      continue;
    }
    auto timeUnsigned = static_cast<uint32_t>(time.tv_sec);
    if (timeUnsigned > fromUnixSeconds && timeUnsigned + rolloverDiffSeconds < upToUnixSeconds) {
      fileToPackets(file, timeUnsigned, funnel);
    }
  }
}

void PersistentTmStore::pathToTod(const std::filesystem::path& path, Clock::TimeOfDay_t& tod) {
  auto pathStr = path.string();
  size_t splitChar = pathStr.find('_');
  auto timeOnlyStr = pathStr.substr(splitChar);
  sscanf(timeOnlyStr.data(),
         "%04" SCNu32 "-%02" SCNu32 "-%02" SCNu32 "T%02" SCNu32 "-%02" SCNu32 "-%02" SCNu32 "Z",
         &tod.year, &tod.month, &tod.day, &tod.hour, &tod.minute, &tod.second);
}

void PersistentTmStore::fileToPackets(const std::filesystem::path& path, uint32_t unixStamp,
                                      TmFunnelBase& funnel) {
  store_address_t storeId;
  TmTcMessage message;
  size_t size = std::filesystem::file_size(path);
  if (size < 6) {
    // Can't even read the CCSDS header
    return;
  }
  std::ifstream ifile(path, std::ios::binary);
  ifile.read(reinterpret_cast<char*>(fileBuf.data()), static_cast<std::streamsize>(size));
  size_t currentIdx = 0;
  while (currentIdx < size) {
    PusTmReader reader(&timeReader, fileBuf.data(), fileBuf.size());
    // CRC check to fully ensure this is a valid TM
    ReturnValue_t result = reader.parseDataWithCrcCheck();
    if (result == returnvalue::OK) {
      result = tmStore.addData(&storeId, fileBuf.data() + currentIdx, reader.getFullPacketLen());
      if (result != returnvalue::OK) {
        continue;
      }
      funnel.sendPacketToDestinations(storeId, message, fileBuf.data() + currentIdx,
                                      reader.getFullPacketLen());
      currentIdx += reader.getFullPacketLen();
    } else {
      sif::error << "Parsing of PUS TM failed with code " << result << std::endl;
      triggerEvent(POSSIBLE_FILE_CORRUPTION, result, unixStamp);
      // Stop for now, do not really know where to continue and we do not trust the file anymore.
      break;
    }
  }
}