file_ipc_stream_writer.hpp

// file_ipc_stream_writer.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 <filesystem>
#include <fstream>
#include <mutex>
#include <chrono>
#include <atomic>
// === third-party includes (if any) ===
#include <google/protobuf/util/json_util.h>
#include <glog/logging.h>
#include <turbojpeg.h>
// === local includes (if any) ===
#include "../ipc_stream_writer.hpp"
#include "file_ipc_configuration.hpp"
 
namespace presage::smartspectra::file_ipc {

template<container::settings::IntegrationMode TIntegrationMode>
class FileIpcStreamWriter : public ipc::IpcStreamWriter<TIntegrationMode> {
public:
    using Base = ipc::IpcStreamWriter<TIntegrationMode>;
    using typename Base::BackgroundContainer;
 
    explicit FileIpcStreamWriter(const FileIpcConfiguration& configuration);
    virtual ~FileIpcStreamWriter();
 
    // Implement base interface
    absl::Status WriteStatus(const presage::physiology::StatusValue& status) override;
    absl::Status WriteCoreMetrics(
        const presage::physiology::MetricsBuffer& metrics,
        int64_t timestamp_us
    ) override;
    absl::Status WriteEdgeMetrics(const presage::physiology::Metrics& metrics, int64_t timestamp_us) override;
    absl::Status WriteFrame(const cv::Mat& frame, int64_t timestamp_us) override;
    void Close() override;
    bool IsOperational() const override;
    absl::Status Flush() override;
 
protected:
    absl::Status Initialize();

    absl::Status WriteJsonLine(std::ofstream& stream, const std::string& json_data);

    static std::string BuildEnvelope(
        const std::string& type,
        const std::string& json_payload,
        int64_t timestamp_us
    );

    static std::string BuildStatusPayload(const presage::physiology::StatusValue& status);

    static std::string JsonEscape(const std::string& value);

    static const google::protobuf::util::JsonPrintOptions& GetJsonOptions();

    absl::Status CompressFrameWithTurboJpeg(const cv::Mat& frame, std::vector<unsigned char>& jpeg_data);
 
private:
    FileIpcConfiguration configuration_;
    std::filesystem::path base_directory_;
 
    // Output streams
    std::ofstream core_metrics_stream_;
    std::ofstream edge_metrics_stream_;
    std::ofstream status_stream_;
 
    // Stream buffers to prevent shared buffer memory corruption
    std::vector<char> core_metrics_buffer_;
    std::vector<char> edge_metrics_buffer_;
    std::vector<char> status_buffer_;
 
    // Frame management
    std::filesystem::path frames_dir_;
 
    // Thread safety
    mutable std::mutex write_mutex_;
    std::mutex frame_mutex_;
 
    // Error handling
    std::atomic<bool> is_operational_{false};
 
    // Recording state management
    std::filesystem::path recording_state_path_;
    std::filesystem::path recording_command_path_;
 
    // TurboJPEG compressor handle
    tjhandle turbo_jpeg_compressor_ = nullptr;
};
 
} // namespace presage::smartspectra::file_ipc