stargazer/tier3/

mod.rs

//! Tier 3: deep D-STAR protocol connections and voice capture.
//!
//! Establishes full D-STAR protocol connections (`DPlus`, `DExtra`, DCS) to
//! reflectors via the `dstar-gateway` crate's `AsyncSession` API. Captures
//! voice streams including:
//!
//! - D-STAR header metadata (callsign, suffix, UR call, RPT fields)
//! - AMBE voice frames decoded to PCM via `mbelib-rs`, then encoded to MP3
//! - Slow data (D-STAR text messages, DPRS position reports)
//!
//! The capture pipeline for each stream:
//!
//! ```text
//! AsyncSession<P> -> VoiceStart (header)
//!                 -> VoiceFrame x N (AMBE + slow data)
//!                 -> VoiceEnd
//!                     |
//!             mbelib-rs: AMBE -> PCM
//!                     |
//!             mp3lame-encoder: PCM -> MP3
//!                     |
//!             Postgres: streams table (metadata + audio_mp3)
//! ```
//!
//! Connections are managed by a priority queue: user-pinned reflectors take
//! precedence, followed by auto-promoted reflectors ranked by activity score.
//! Idle sessions (no voice activity beyond the configured timeout) are
//! disconnected to free slots for more active reflectors.
//!
//! # Module layout
//!
//! - [`decoder`] — AMBE-to-PCM-to-MP3 audio pipeline using `mbelib-rs` and
//!   `mp3lame-encoder`.
//! - [`capture`] — Per-stream state tracker (`StreamCapture`) and
//!   cross-stream dispatcher (`CaptureManager`).
//!
//! # Current status
//!
//! The audio decode pipeline and capture-state machinery are complete and
//! unit-tested. The remaining orchestration work — maintaining a pool of
//! `AsyncSession` tasks keyed off the Tier 1/Tier 2 reflector scores — is
//! left to a follow-up. The public [`run`] entry point currently logs its
//! configuration and blocks forever; [`process_completed_stream`] is the
//! production-grade finalizer that the future orchestrator will wire in.

pub(crate) mod capture;
pub(crate) mod decoder;

use chrono::Utc;
use sqlx::PgPool;

use crate::config::{AudioConfig, Tier3Config};
use crate::db::streams::{self, AudioUpdate, NewStream};

use capture::StreamCapture;

/// Runs the Tier 3 voice capture orchestrator.
///
/// Manages a pool of D-STAR protocol connections, capturing voice streams and
/// writing decoded audio with metadata to `PostgreSQL`. Runs until cancelled.
///
/// # Current behavior
///
/// This is a placeholder: the connection-pool management layer that
/// drives `AsyncSession<P>` is not yet implemented. For now the
/// function logs its configuration and then blocks on a never-completing
/// future so the main task-supervisor loop treats it as a well-behaved
/// long-running task.
///
/// # Errors
///
/// Returns an error if a fatal, non-retryable failure occurs (e.g., the
/// database pool is closed). Today, the stub path cannot fail.
pub(crate) async fn run(
    config: Tier3Config,
    audio_config: AudioConfig,
    _pool: PgPool,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    tracing::info!(
        max_concurrent_connections = config.max_concurrent_connections,
        idle_disconnect_secs = config.idle_disconnect_secs,
        auto_promote = config.auto_promote,
        dplus_callsign = %config.dplus_callsign,
        audio_format = %audio_config.format,
        mp3_bitrate = audio_config.mp3_bitrate,
        "tier3 voice capture: session-pool orchestration not yet wired — \
         decoder + capture subsystems ready"
    );

    // TODO: spawn the AsyncSession pool, fan `VoiceStart`/`VoiceFrame`/
    // `VoiceEnd` events into `CaptureManager`, and call
    // `process_completed_stream` on each terminated capture.
    std::future::pending::<()>().await;
    Ok(())
}

/// Finalizes one captured voice stream: MP3-encodes the audio and persists
/// the row to `PostgreSQL`.
///
/// Called by the Tier 3 orchestrator when a `VoiceEnd` event fires (or an
/// idle timeout elapses). The orchestrator is stubbed in [`run`] above so
/// the function is unreferenced at build time today — the follow-up task
/// will wire it into the session pool.
///
/// Two database writes happen:
///
/// 1. [`streams::insert_stream`] — creates the row with header metadata.
/// 2. [`streams::update_audio`] — fills in the MP3 blob, frame count, end
///    timestamp, and decoded slow-data fields (text, DPRS lat/lon).
///
/// The split into insert + update mirrors the schema lifecycle documented
/// in `db/streams.rs`: the row is visible to the HTTP API as soon as the
/// header arrives, and its `audio_mp3` field transitions from NULL to a
/// populated blob when encoding completes.
///
/// # Errors
///
/// - AMBE/MP3 decode errors ([`decoder::DecodeError`]) are wrapped in the
///   returned boxed error and logged at `warn` level — one bad stream
///   should not abort the capture loop.
/// - SQL errors from the insert or update are bubbled up so the caller can
///   distinguish transient database issues from bad audio.
#[expect(
    dead_code,
    reason = "wired into the session pool in the follow-up orchestrator task"
)]
pub(crate) async fn process_completed_stream(
    capture: StreamCapture,
    pool: &PgPool,
    mp3_bitrate: u32,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let frame_count = capture.frame_count();
    if frame_count == 0 {
        // Empty capture: nothing to encode. Drop silently — the header
        // alone is insufficient to justify a row (the HTTP API would show
        // a zero-duration "transmission" that serves no purpose).
        tracing::debug!(
            reflector = %capture.reflector,
            module = %capture.module,
            stream_id = capture.stream_id,
            "stream ended with zero voice frames — dropping"
        );
        return Ok(());
    }

    // Encode AMBE frames to MP3. This is CPU-bound but short (typically
    // a few milliseconds even for a 30-second transmission); inline is
    // fine rather than `spawn_blocking`.
    let mp3_bytes = decoder::decode_to_mp3(&capture.ambe_frames, mp3_bitrate).map_err(|e| {
        tracing::warn!(
            reflector = %capture.reflector,
            module = %capture.module,
            stream_id = capture.stream_id,
            frame_count,
            error = %e,
            "failed to encode captured stream to MP3"
        );
        e
    })?;

    // Insert the row with header metadata. `audio_mp3` starts NULL; the
    // second query (`update_audio`) fills it in.
    let new_stream = NewStream {
        reflector: &capture.reflector,
        module: &capture.module,
        protocol: &capture.protocol,
        stream_id: i32::from(capture.stream_id),
        callsign: &capture.callsign,
        suffix: capture.suffix.as_deref(),
        ur_call: capture.ur_call.as_deref(),
        started_at: capture.started_at,
    };
    let row_id = streams::insert_stream(pool, &new_stream).await?;

    // Populate audio + metadata. frame_count is usize on our side but
    // Postgres stores INTEGER (i32); saturate on overflow to avoid a panic
    // on pathologically long (>2.1B-frame) transmissions.
    let frame_count_i32 = i32::try_from(frame_count).unwrap_or(i32::MAX);
    let audio_update = AudioUpdate {
        audio_mp3: &mp3_bytes,
        ended_at: Utc::now(),
        frame_count: frame_count_i32,
        dstar_text: capture.dstar_text.as_deref(),
        dprs_lat: capture.dprs_lat,
        dprs_lon: capture.dprs_lon,
    };
    streams::update_audio(pool, row_id, &audio_update).await?;

    tracing::info!(
        row_id,
        reflector = %capture.reflector,
        module = %capture.module,
        protocol = %capture.protocol,
        stream_id = capture.stream_id,
        callsign = %capture.callsign,
        frame_count,
        mp3_bytes = mp3_bytes.len(),
        has_text = capture.dstar_text.is_some(),
        has_dprs = capture.dprs_lat.is_some(),
        "voice stream captured and stored"
    );

    Ok(())
}