file_ipc_stream_writer_impl.hpp

// file_ipc_stream_writer_impl.hpp
// Created by Greg on 9/26/2025.
// Copyright (C) 2025 Presage Security, Inc.
//
// SPDX-License-Identifier: LGPL-3.0-or-later
 
#pragma once
 
// === standard library includes (if any) ===
#include <iomanip>
#include <sstream>
#include <vector>
#include <algorithm>
#include <limits>
// === third-party includes (if any) ===
#include <absl/strings/str_cat.h>
#include <physiology/modules/messages/status.h>
// === local includes (if any) ===
#include "file_ipc_stream_writer.hpp"
 
namespace presage::smartspectra::file_ipc {
 
template<container::settings::IntegrationMode TIntegrationMode>
FileIpcStreamWriter<TIntegrationMode>::FileIpcStreamWriter(const FileIpcConfiguration& configuration)
    : configuration_(configuration), base_directory_(configuration.base_directory) {
    // Initialize TurboJPEG compressor
    turbo_jpeg_compressor_ = tjInitCompress();
    if (!turbo_jpeg_compressor_) {
        LOG(ERROR) << "Failed to initialize TurboJPEG compressor";
    }
 
    auto status = Initialize();
    if (!status.ok()) {
        LOG(ERROR) << "Failed to initialize FileIpcStreamWriter: " << status.message();
    }
}
 
template<container::settings::IntegrationMode TIntegrationMode>
FileIpcStreamWriter<TIntegrationMode>::~FileIpcStreamWriter() {
    Close();
 
    // Clean up TurboJPEG compressor
    if (turbo_jpeg_compressor_) {
        tjDestroy(turbo_jpeg_compressor_);
        turbo_jpeg_compressor_ = nullptr;
    }
}
 
template<container::settings::IntegrationMode TIntegrationMode>
absl::Status FileIpcStreamWriter<TIntegrationMode>::Initialize() {
    std::lock_guard<std::mutex> lock(write_mutex_);
 
    // Create base directory if needed
    if (configuration_.create_if_missing) {
        std::error_code ec;
        std::filesystem::create_directories(base_directory_, ec);
        if (ec) {
            return absl::InternalError(
                absl::StrCat("Failed to create base directory ", base_directory_.string(), ": ", ec.message())
            );
        }
    }
 
    if (!std::filesystem::exists(base_directory_)) {
        return absl::NotFoundError(absl::StrCat("Base directory does not exist: ", base_directory_.string()));
    }
 
    // Create frames directory
    frames_dir_ = base_directory_ / "frames";
    if (configuration_.create_if_missing) {
        std::error_code ec;
        std::filesystem::create_directories(frames_dir_, ec);
        if (ec) {
            return absl::InternalError(
                absl::StrCat("Failed to create frames directory: ", ec.message())
            );
        }
    }
 
    // Open stream files
    auto open_mode = configuration_.truncate_on_init
        ? (std::ios::out | std::ios::trunc)
        : (std::ios::out | std::ios::app);
 
    core_metrics_stream_.open(base_directory_ / "core_metrics.jsonl", open_mode);
    edge_metrics_stream_.open(base_directory_ / "edge_metrics.jsonl", open_mode);
    status_stream_.open(base_directory_ / "status_codes.jsonl", open_mode);
 
    if (!core_metrics_stream_.is_open() || !edge_metrics_stream_.is_open() || !status_stream_.is_open()) {
        return absl::InternalError("Failed to open one or more output streams");
    }
 
    // Set buffer size if specified
    if (configuration_.io_buffer_size > 0) {
        // Create separate buffers for each stream to avoid memory corruption
        core_metrics_buffer_ = std::vector<char>(configuration_.io_buffer_size);
        edge_metrics_buffer_ = std::vector<char>(configuration_.io_buffer_size);
        status_buffer_ = std::vector<char>(configuration_.io_buffer_size);
 
        core_metrics_stream_.rdbuf()->pubsetbuf(core_metrics_buffer_.data(), core_metrics_buffer_.size());
        edge_metrics_stream_.rdbuf()->pubsetbuf(edge_metrics_buffer_.data(), edge_metrics_buffer_.size());
        status_stream_.rdbuf()->pubsetbuf(status_buffer_.data(), status_buffer_.size());
    }
 
    // Setup recording state files
    recording_state_path_ = base_directory_ / "recording_state.json";
    recording_command_path_ = base_directory_ / "recording_command.json";
 
    // Initialize recording state
    std::ofstream recording_state(recording_state_path_, std::ios::out | std::ios::trunc);
    if (recording_state.is_open()) {
        recording_state << R"({"recording":false,"timestamp":)"
                        << this->GetCurrentTimestampUs() << "}\n";
        recording_state.close();
    }
 
    // Remove any stale command file
    if (std::filesystem::exists(recording_command_path_)) {
        std::error_code ec;
        std::filesystem::remove(recording_command_path_, ec);
        if (ec && configuration_.enable_debug_logging) {
            LOG(WARNING) << "Failed to remove stale command file: " << ec.message();
        }
    }
 
    is_operational_ = true;
    return absl::OkStatus();
}
 
template<container::settings::IntegrationMode TIntegrationMode>
absl::Status FileIpcStreamWriter<TIntegrationMode>::WriteStatus(const presage::physiology::StatusValue& status) {
    if (!is_operational_) {
        return absl::FailedPreconditionError("Writer is not operational");
    }
 
    std::lock_guard<std::mutex> lock(write_mutex_);
    std::string payload = BuildEnvelope("status", BuildStatusPayload(status), status.timestamp());
    return WriteJsonLine(status_stream_, payload);
}
 
template<container::settings::IntegrationMode TIntegrationMode>
absl::Status FileIpcStreamWriter<TIntegrationMode>::WriteCoreMetrics(
    const presage::physiology::MetricsBuffer& metrics,
    int64_t timestamp_us
) {
    if (!is_operational_) {
        return absl::FailedPreconditionError("Writer is not operational");
    }
 
    std::string json;
    auto status = (google::protobuf::util::MessageToJsonString(metrics, &json, GetJsonOptions()));
    if (!status.ok()) {
        return absl::InternalError(
            absl::StrCat("Failed to serialize MetricsBuffer: ", status.message().as_string())
        );
    }
 
    std::lock_guard<std::mutex> lock(write_mutex_);
    std::string payload = BuildEnvelope("core_metrics", json, timestamp_us);
    return WriteJsonLine(core_metrics_stream_, payload);
}
 
template<container::settings::IntegrationMode TIntegrationMode>
absl::Status FileIpcStreamWriter<TIntegrationMode>::WriteEdgeMetrics(
    const presage::physiology::Metrics& metrics, int64_t timestamp_us
) {
    if (!is_operational_) {
        return absl::FailedPreconditionError("Writer is not operational");
    }
 
    std::string json;
    auto status = google::protobuf::util::MessageToJsonString(metrics, &json, GetJsonOptions());
    if (!status.ok()) {
        return absl::InternalError(
            absl::StrCat("Failed to serialize Metrics: ", status.message().as_string())
        );
    }
 
    std::lock_guard<std::mutex> lock(write_mutex_);
    std::string payload = BuildEnvelope("edge_metrics", json, timestamp_us);
    return WriteJsonLine(edge_metrics_stream_, payload);
}
 
template<container::settings::IntegrationMode TIntegrationMode>
absl::Status FileIpcStreamWriter<TIntegrationMode>::WriteFrame(const cv::Mat& frame, int64_t timestamp_us) {
    if (!is_operational_) {
        return absl::FailedPreconditionError("Writer is not operational");
    }
 
    if (frame.empty()) {
        return absl::InvalidArgumentError("Frame is empty");
    }
 
    if (!turbo_jpeg_compressor_) {
        return absl::FailedPreconditionError("TurboJPEG compressor not initialized");
    }
 
    // Use fixed paths like hud_frame_streamer - only maintain latest.jpg and latest.json
    std::lock_guard<std::mutex> lock(frame_mutex_);
 
    std::filesystem::path frame_path = frames_dir_ / "latest.jpg";
    std::filesystem::path temp_frame_path = frames_dir_ / "latest.jpg.tmp";
    std::filesystem::path metadata_path = frames_dir_ / "latest.json";
    std::filesystem::path temp_metadata_path = frames_dir_ / "latest.json.tmp";
 
    // Compress frame using TurboJPEG
    std::vector<unsigned char> jpeg_data;
    auto compress_status = CompressFrameWithTurboJpeg(frame, jpeg_data);
    if (!compress_status.ok()) {
        return compress_status;
    }
 
    // Write compressed frame to temporary file first (atomic write)
    std::ofstream frame_stream(temp_frame_path, std::ios::binary | std::ios::trunc);
    if (!frame_stream.is_open()) {
        return absl::InternalError(absl::StrCat("Failed to open ", temp_frame_path.string(), " for writing"));
    }
 
    frame_stream.write(reinterpret_cast<const char*>(jpeg_data.data()), jpeg_data.size());
    frame_stream.close();
 
    if (!frame_stream) {
        return absl::InternalError(absl::StrCat("Failed to write frame to ", temp_frame_path.string()));
    }
 
    // Atomically rename frame to final path
    std::error_code ec;
    // Remove existing file first to avoid issues on some filesystems
    std::filesystem::remove(frame_path, ec);
    ec.clear();
    std::filesystem::rename(temp_frame_path, frame_path, ec);
    if (ec) {
        return absl::InternalError(absl::StrCat("Failed to rename frame file: ", ec.message()));
    }
 
    // Write metadata JSON file
    std::ostringstream metadata;
    metadata << '{'
             << "\"timestamp\":" << timestamp_us << ','
             << "\"width\":" << frame.cols << ','
             << "\"height\":" << frame.rows << ','
             << "\"channels\":" << frame.channels()
             << '}';
 
    std::ofstream metadata_stream(temp_metadata_path, std::ios::out | std::ios::trunc);
    if (!metadata_stream.is_open()) {
        // Log warning but don't fail - frame is already written
        LOG(WARNING) << "Failed to open metadata file: " << temp_metadata_path;
        return absl::OkStatus();
    }
 
    metadata_stream << metadata.str() << '\n';
    metadata_stream.close();
 
    if (!metadata_stream) {
        LOG(WARNING) << "Failed to write metadata file: " << temp_metadata_path;
        return absl::OkStatus();
    }
 
    // Atomically rename metadata to final path
    std::filesystem::remove(metadata_path, ec);
    ec.clear();
    std::filesystem::rename(temp_metadata_path, metadata_path, ec);
    if (ec) {
        LOG(WARNING) << "Failed to rename metadata file: " << ec.message();
    }
 
    return absl::OkStatus();
}
 
template<container::settings::IntegrationMode TIntegrationMode>
void FileIpcStreamWriter<TIntegrationMode>::Close() {
    is_operational_ = false;
    std::lock_guard<std::mutex> lock(write_mutex_);
 
    if (core_metrics_stream_.is_open()) {
        core_metrics_stream_.flush();
        core_metrics_stream_.close();
    }
    if (edge_metrics_stream_.is_open()) {
        edge_metrics_stream_.flush();
        edge_metrics_stream_.close();
    }
    if (status_stream_.is_open()) {
        status_stream_.flush();
        status_stream_.close();
    }
}
 
template<container::settings::IntegrationMode TIntegrationMode>
bool FileIpcStreamWriter<TIntegrationMode>::IsOperational() const {
    return is_operational_;
}
 
template<container::settings::IntegrationMode TIntegrationMode>
absl::Status FileIpcStreamWriter<TIntegrationMode>::Flush() {
    if (!is_operational_) {
        return absl::FailedPreconditionError("Writer is not operational");
    }
 
    std::lock_guard<std::mutex> lock(write_mutex_);
    core_metrics_stream_.flush();
    edge_metrics_stream_.flush();
    status_stream_.flush();
 
    if (core_metrics_stream_.fail() || edge_metrics_stream_.fail() || status_stream_.fail()) {
        return absl::InternalError("Failed to flush one or more streams");
    }
 
    return absl::OkStatus();
}
 
template<container::settings::IntegrationMode TIntegrationMode>
absl::Status FileIpcStreamWriter<TIntegrationMode>::WriteJsonLine(std::ofstream& stream, const std::string& json_data) {
    if (!stream.is_open()) {
        return absl::FailedPreconditionError("Stream is not open");
    }
 
    stream << json_data << '\n';
    stream.flush();
 
    if (stream.fail()) {
        return absl::InternalError("Failed to write to stream");
    }
 
    return absl::OkStatus();
}
 
template<container::settings::IntegrationMode TIntegrationMode>
std::string FileIpcStreamWriter<TIntegrationMode>::BuildEnvelope(
    const std::string& type,
    const std::string& json_payload,
    int64_t timestamp_us
) {
    std::ostringstream envelope;
    envelope << '{'
             << "\"type\":\"" << type << "\","
             << "\"timestamp\":" << timestamp_us << ','
             << "\"payload\":" << json_payload
             << '}';
    return envelope.str();
}
 
template<container::settings::IntegrationMode TIntegrationMode>
std::string FileIpcStreamWriter<TIntegrationMode>::BuildStatusPayload(const presage::physiology::StatusValue& status) {
    std::ostringstream payload;
    payload << '{'
            << "\"value\":" << status.value() << ','
            << "\"timestamp\":" << status.timestamp() << ','
            << "\"description\":\""
            << JsonEscape(presage::physiology::GetStatusDescription(status.value())) << "\","
            << "\"hint\":\""
            << JsonEscape(presage::physiology::GetStatusHint(status.value())) << "\""
            << '}';
    return payload.str();
}
 
template<container::settings::IntegrationMode TIntegrationMode>
std::string FileIpcStreamWriter<TIntegrationMode>::JsonEscape(const std::string& value) {
    std::string escaped;
    escaped.reserve(value.size() * 2);
    for (unsigned char ch : value) {
        switch (ch) {
            case '\\': escaped += "\\\\"; break;
            case '\"': escaped += "\\\""; break;
            case '\n': escaped += "\\n"; break;
            case '\r': escaped += "\\r"; break;
            case '\t': escaped += "\\t"; break;
            case '\b': escaped += "\\b"; break;
            case '\f': escaped += "\\f"; break;
            default:
                if (ch < 0x20) {
                    escaped += "\\u00";
                    escaped += "0123456789abcdef"[ch >> 4];
                    escaped += "0123456789abcdef"[ch & 0x0f];
                } else {
                    escaped.push_back(static_cast<char>(ch));
                }
                break;
        }
    }
    return escaped;
}
 
template<container::settings::IntegrationMode TIntegrationMode>
const google::protobuf::util::JsonPrintOptions& FileIpcStreamWriter<TIntegrationMode>::GetJsonOptions() {
    static google::protobuf::util::JsonPrintOptions options = []() {
        google::protobuf::util::JsonPrintOptions opts;
        opts.preserve_proto_field_names = true;
        opts.add_whitespace = false;
        opts.always_print_primitive_fields = false;
        return opts;
    }();
    return options;
}
 
template<container::settings::IntegrationMode TIntegrationMode>
absl::Status FileIpcStreamWriter<TIntegrationMode>::CompressFrameWithTurboJpeg(
    const cv::Mat& frame,
    std::vector<unsigned char>& jpeg_data
) {
    if (!turbo_jpeg_compressor_) {
        return absl::FailedPreconditionError("TurboJPEG compressor not initialized");
    }
 
    if (frame.depth() != CV_8U) {
        return absl::InvalidArgumentError("Frame must have 8-bit depth for JPEG compression");
    }
 
    // Determine pixel format based on number of channels
    int pixel_format = TJPF_BGR;
    switch (frame.channels()) {
        case 1:
            pixel_format = TJPF_GRAY;
            break;
        case 3:
            pixel_format = TJPF_BGR;
            break;
        case 4:
            pixel_format = TJPF_BGRA;
            break;
        default:
            return absl::InvalidArgumentError(
                absl::StrCat("Unsupported number of channels: ", frame.channels())
            );
    }
 
    // Prepare compression parameters
    unsigned char* raw_jpeg = nullptr;
    unsigned long jpeg_size = 0;
 
    const int subsample = TJSAMP_420;  // 4:2:0 subsampling for better compression
    const int flags = TJFLAG_FASTUPSAMPLE | TJFLAG_FASTDCT;  // Fast compression flags
 
    // Check that frame stride doesn't exceed TurboJPEG limits
    if (frame.step > static_cast<size_t>(std::numeric_limits<int>::max())) {
        return absl::InvalidArgumentError("Frame stride exceeds TurboJPEG limits");
    }
    const int pitch = frame.step ? static_cast<int>(frame.step) : 0;
 
    // Perform compression
    int compress_result = tjCompress2(
        turbo_jpeg_compressor_,
        frame.data,
        frame.cols,
        pitch,
        frame.rows,
        pixel_format,
        &raw_jpeg,
        &jpeg_size,
        subsample,
        configuration_.jpeg_quality,  // Use configured JPEG quality
        flags
    );
 
    if (compress_result != 0) {
        // Clean up if allocated
        if (raw_jpeg) {
            tjFree(raw_jpeg);
        }
 
        // Get error message from TurboJPEG
        const char* error_msg = tjGetErrorStr2(turbo_jpeg_compressor_);
        std::string error_text = error_msg ? error_msg : "unknown error";
        return absl::InternalError(absl::StrCat("TurboJPEG compression failed: ", error_text));
    }
 
    // Copy compressed data to output vector
    jpeg_data.clear();
    jpeg_data.reserve(jpeg_size);
    jpeg_data.assign(raw_jpeg, raw_jpeg + jpeg_size);
 
    // Free TurboJPEG buffer
    tjFree(raw_jpeg);
 
    if (configuration_.enable_debug_logging) {
        LOG(INFO) << "Compressed frame from " << (frame.total() * frame.elemSize())
                  << " bytes to " << jpeg_size << " bytes (quality="
                  << configuration_.jpeg_quality << ")";
    }
 
    return absl::OkStatus();
}
 
} // namespace presage::smartspectra::file_ipc