redis_ipc_stream_writer_impl.hpp

// redis_ipc_stream_writer_impl.hpp
// Created by Greg on 1/30/25.
// Copyright (C) 2025 Presage Security, Inc.
//
// SPDX-License-Identifier: LGPL-3.0-or-later
 
#pragma once
 
#include "redis_ipc_stream_writer.hpp"
 
#include <sstream>
#include <glog/logging.h>
#include <absl/strings/escaping.h>
#include <google/protobuf/util/json_util.h>
#include <mediapipe/framework/port/ret_check.h>
#include "../json_envelope.hpp"
#include "../turbojpeg_compressor.hpp"
 
namespace presage::smartspectra::redis_ipc {
 
template<container::settings::IntegrationMode TIntegrationMode>
RedisIpcStreamWriter<TIntegrationMode>::RedisIpcStreamWriter(const RedisIpcConfig& configuration)
    : configuration_(configuration) {
    redis_client_.SetResponseLimits(
        configuration_.max_bulk_string_bytes,
        configuration_.max_array_elements
    );
    auto status = Initialize();
    if (!status.ok()) {
        LOG(ERROR) << "Failed to initialize RedisIpcStreamWriter: " << status.message();
    }
}
 
template<container::settings::IntegrationMode TIntegrationMode>
RedisIpcStreamWriter<TIntegrationMode>::~RedisIpcStreamWriter() {
    Close();
}
 
template<container::settings::IntegrationMode TIntegrationMode>
absl::Status RedisIpcStreamWriter<TIntegrationMode>::Initialize() {
    std::lock_guard<std::mutex> lock(write_mutex_);
 
    // Connect to Redis
    if (!redis_client_.Connect(
        configuration_.host,
        configuration_.port,
        configuration_.connection_timeout
    )) {
        return absl::InternalError(
            "Failed to connect to Redis: " + redis_client_.last_error()
        );
    }
 
    // Clear recording commands queue
    if (!redis_client_.Del(BuildKey(configuration_.keys.command_queue))) {
        if (configuration_.enable_debug_logging) {
            LOG(WARNING) << "Failed to clear Redis recording commands: "
                        << redis_client_.last_error();
        }
    }
 
    // Initialize recording state
    const std::string initial_state = R"({"recording":false,"timestamp":)" +
                                      std::to_string(this->GetCurrentTimestampUs()) + "}";
 
    if (!redis_client_.Set(BuildKey(configuration_.keys.recording_state), initial_state)) {
        if (configuration_.enable_debug_logging) {
            LOG(WARNING) << "Failed to set initial Redis recording state: "
                        << redis_client_.last_error();
        }
    }
 
    // Publish initial state
    if (!redis_client_.Publish(BuildChannel("recording_state"), initial_state)) {
        if (configuration_.enable_debug_logging) {
            LOG(WARNING) << "Failed to publish initial recording state: "
                        << redis_client_.last_error();
        }
    }
 
    is_operational_ = true;
    return absl::OkStatus();
}
 
template<container::settings::IntegrationMode TIntegrationMode>
absl::Status RedisIpcStreamWriter<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_);
 
    const std::string payload = BuildJsonEnvelope(
        "status",
        BuildStatusPayload(status),
        status.timestamp()
    );
 
    if (!redis_client_.Publish(BuildChannel(configuration_.channels.status), payload)) {
        if (publish_error_count_++ < kMaxErrorLogs) {
            LOG(WARNING) << "Failed to publish status to Redis: "
                        << redis_client_.last_error();
        }
    }
 
    return absl::OkStatus();
}
 
template<container::settings::IntegrationMode TIntegrationMode>
absl::Status RedisIpcStreamWriter<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("Failed to serialize MetricsBuffer: " + status.message().as_string());
    }
 
    std::lock_guard<std::mutex> lock(write_mutex_);
 
    const std::string payload = BuildJsonEnvelope("core_metrics", json, timestamp_us);
 
    if (!redis_client_.Publish(BuildChannel(configuration_.channels.core_metrics), payload)) {
        if (publish_error_count_++ < kMaxErrorLogs) {
            LOG(WARNING) << "Failed to publish core metrics to Redis: "
                        << redis_client_.last_error();
        }
    }
 
    return absl::OkStatus();
}
 
template<container::settings::IntegrationMode TIntegrationMode>
absl::Status RedisIpcStreamWriter<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("Failed to serialize Metrics: " + status.message().as_string());
    }
 
    std::lock_guard<std::mutex> lock(write_mutex_);
 
    const std::string payload = BuildJsonEnvelope("edge_metrics", json, timestamp_us);
 
    if (!redis_client_.Publish(BuildChannel(configuration_.channels.edge_metrics), payload)) {
        if (publish_error_count_++ < kMaxErrorLogs) {
            LOG(WARNING) << "Failed to publish edge metrics to Redis: "
                        << redis_client_.last_error();
        }
    }
 
    return absl::OkStatus();
}
 
template<container::settings::IntegrationMode TIntegrationMode>
absl::Status RedisIpcStreamWriter<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");
    }
 
    // Create compressor for this frame
    MP_ASSIGN_OR_RETURN(auto compressor, compression::TurboJpegCompressor::Create());
 
    // Compress frame
    std::vector<unsigned char> jpeg_data;
    MP_RETURN_IF_ERROR(compressor.Compress(frame, 90, jpeg_data));
 
    // Base64 encode
    std::string base64;
    absl::Base64Escape(absl::string_view(
        reinterpret_cast<const char*>(jpeg_data.data()),
        jpeg_data.size()
    ), &base64);
 
    std::lock_guard<std::mutex> lock(write_mutex_);
 
    // Build payload with metadata
    std::ostringstream payload;
    payload << '{'
            << "\"timestamp\":" << timestamp_us << ','
            << "\"width\":" << frame.cols << ','
            << "\"height\":" << frame.rows << ','
            << "\"channels\":" << frame.channels() << ','
            << "\"jpeg_base64\":\"" << base64 << "\""
            << '}';
 
    const std::string payload_str = payload.str();
 
    // Publish to channel for real-time streaming
    if (!redis_client_.Publish(BuildChannel(configuration_.channels.hud_frame), payload_str)) {
        if (publish_error_count_++ < kMaxErrorLogs) {
            LOG(WARNING) << "Failed to publish frame to Redis: "
                        << redis_client_.last_error();
        }
    }
 
    return absl::OkStatus();
}
 
template<container::settings::IntegrationMode TIntegrationMode>
void RedisIpcStreamWriter<TIntegrationMode>::Close() {
    is_operational_ = false;
    std::lock_guard<std::mutex> lock(write_mutex_);
    redis_client_.Close();
}
 
template<container::settings::IntegrationMode TIntegrationMode>
bool RedisIpcStreamWriter<TIntegrationMode>::IsOperational() const {
    return is_operational_;
}
 
template<container::settings::IntegrationMode TIntegrationMode>
absl::Status RedisIpcStreamWriter<TIntegrationMode>::Flush() {
    // Redis pub/sub is fire-and-forget, no buffering to flush
    return absl::OkStatus();
}
 
template<container::settings::IntegrationMode TIntegrationMode>
absl::Status RedisIpcStreamWriter<TIntegrationMode>::PublishRecordingState(bool recording) {
    if (!is_operational_) {
        return absl::FailedPreconditionError("Writer is not operational");
    }
 
    std::lock_guard<std::mutex> lock(write_mutex_);
 
    // Build recording state JSON
    const std::string state = R"({"recording":)" + std::string(recording ? "true" : "false") +
                              R"(,"timestamp":)" + std::to_string(this->GetCurrentTimestampUs()) + "}";
 
    // Publish to recording_state channel
    if (!redis_client_.Publish(BuildChannel("recording_state"), state)) {
        if (configuration_.enable_debug_logging) {
            LOG(WARNING) << "Failed to publish recording state: " << redis_client_.last_error();
        }
        return absl::InternalError("Failed to publish recording state: " + redis_client_.last_error());
    }
 
    // Also update recording_state key for persistence
    if (!redis_client_.Set(BuildKey(configuration_.keys.recording_state), state)) {
        if (configuration_.enable_debug_logging) {
            LOG(WARNING) << "Failed to set recording state key: " << redis_client_.last_error();
        }
    }
 
    return absl::OkStatus();
}
 
template<container::settings::IntegrationMode TIntegrationMode>
std::string RedisIpcStreamWriter<TIntegrationMode>::BuildKey(const std::string& suffix) const {
    if (configuration_.key_prefix.empty()) {
        return suffix;
    }
    return configuration_.key_prefix + ":" + suffix;
}
 
template<container::settings::IntegrationMode TIntegrationMode>
std::string RedisIpcStreamWriter<TIntegrationMode>::BuildChannel(const std::string& suffix) const {
    return BuildKey(suffix);
}
 
} // namespace presage::smartspectra::redis_ipc