hypercall_api/

push_service.rs

//! Web Push notification service.
//!
//! Stores push subscriptions per wallet in Postgres and delivers
//! notifications via the Web Push protocol (VAPID-signed, RFC 8291).
//!
//! Subscriptions are cached in memory (DashMap) to avoid a DB query
//! on every fill. The cache is invalidated on subscribe/unsubscribe.
//!
//! Push delivery is decoupled from the event forwarder via a bounded
//! mpsc channel. A single background task drains the queue and sends
//! up to MAX_CONCURRENT_SENDS push messages in parallel.

use anyhow::{Context, Result};
use dashmap::DashMap;
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use tokio::sync::mpsc;
use tracing::{debug, info, warn};
use web_push::{ContentEncoding, SubscriptionInfo, VapidSignatureBuilder, WebPushMessageBuilder};

use hypercall_db::{
    PushSubscriptionReader, PushSubscriptionRecord, PushSubscriptionWriter,
    UpsertPushSubscriptionInput,
};

// ---------------------------------------------------------------------------
// Notification types and preferences
// ---------------------------------------------------------------------------

/// Notification category. The backend checks the user's preferences
/// before sending a push for each category.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum NotificationType {
    Fill,
    Liquidation,
    Settlement,
}

impl NotificationType {
    fn preference_key(self) -> &'static str {
        match self {
            Self::Fill => "fills",
            Self::Liquidation => "liquidations",
            Self::Settlement => "settlements",
        }
    }
}

/// Default preferences: everything enabled.
pub fn default_preferences() -> serde_json::Value {
    serde_json::json!({
        "fills": true,
        "liquidations": true,
        "settlements": true,
    })
}

/// Check whether a subscription has a given notification type enabled.
/// Missing keys default to enabled. Non-boolean values are treated as
/// disabled (reject malformed preferences rather than silently enabling).
fn is_enabled(preferences: &serde_json::Value, notif_type: NotificationType) -> bool {
    match preferences.get(notif_type.preference_key()) {
        None => true,                            // Key absent = enabled by default
        Some(v) => v.as_bool().unwrap_or(false), // Non-bool = disabled
    }
}

// ---------------------------------------------------------------------------
// Notification payload
// ---------------------------------------------------------------------------

/// JSON payload sent inside the push message. The service worker parses
/// this to decide what to show.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PushPayload {
    pub title: String,
    pub body: String,
    /// Deduplication tag; the browser replaces notifications with the same tag.
    pub tag: String,
    /// Optional deep-link URL to open on click.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub url: Option<String>,
}

/// Internal message sent through the bounded channel from the event
/// forwarder to the background push sender task.
struct PushJob {
    wallet: String,
    notif_type: NotificationType,
    payload: PushPayload,
}

// ---------------------------------------------------------------------------
// Service
// ---------------------------------------------------------------------------

/// Maximum number of wallets to cache. When exceeded, the cache is cleared
/// entirely (simple eviction strategy suitable for this use case since most
/// entries are empty vecs and re-population is cheap).
const CACHE_MAX_ENTRIES: usize = 50_000;

/// Maximum push subscriptions per wallet. Prevents a single wallet from
/// registering thousands of endpoints.
const MAX_SUBSCRIPTIONS_PER_WALLET: i64 = 10;

/// Allowed push endpoint URL prefixes. Rejects anything that isn't a
/// known browser push service to prevent SSRF.
const ALLOWED_PUSH_ENDPOINT_PREFIXES: &[&str] = &[
    "https://fcm.googleapis.com/", // Chrome (Firebase Cloud Messaging)
    "https://updates.push.services.mozilla.com/", // Firefox
    "https://wns.windows.com/",    // Edge (Windows Push Notification Service)
    "https://web.push.apple.com/", // Safari
];

/// Bounded channel capacity for pending push jobs. If the sender task
/// falls behind, new jobs are dropped (best-effort delivery).
const PUSH_QUEUE_CAPACITY: usize = 1_000;

/// Maximum number of push HTTP requests in flight at once.
const MAX_CONCURRENT_SENDS: usize = 16;

pub struct PushNotificationService {
    db: Arc<dyn PushSubscriptionWriter>,
    /// In-memory cache: wallet -> subscriptions.
    /// Wallets with no subscriptions are cached as empty vecs.
    /// Bounded by CACHE_MAX_ENTRIES to prevent unbounded memory growth.
    cache: Arc<DashMap<String, Vec<PushSubscriptionRecord>>>,
    /// Sender half of the push job channel. The background task owns the receiver.
    job_tx: mpsc::Sender<PushJob>,
}

impl PushNotificationService {
    /// Create the service and spawn the background sender task.
    pub fn new(db: Arc<dyn PushSubscriptionWriter>, vapid_private_pem: &[u8]) -> Result<Self> {
        // Validate the PEM key is parseable at startup (fail fast)
        VapidSignatureBuilder::from_pem_no_sub(std::io::Cursor::new(vapid_private_pem))
            .context("VAPID private key is not valid PEM")?;

        let client = Arc::new(reqwest::Client::new());

        let cache: Arc<DashMap<String, Vec<PushSubscriptionRecord>>> = Arc::new(DashMap::new());
        let vapid_key = Arc::new(vapid_private_pem.to_vec());

        let (job_tx, job_rx) = mpsc::channel::<PushJob>(PUSH_QUEUE_CAPACITY);

        // Spawn the background sender task
        let sender_db = db.clone();
        let sender_client = client.clone();
        let sender_cache = cache.clone();
        let sender_key = vapid_key.clone();
        tokio::spawn(push_sender_loop(
            job_rx,
            sender_db,
            sender_client,
            sender_cache,
            sender_key,
        ));

        Ok(Self { db, cache, job_tx })
    }

    // -- Subscription CRUD --------------------------------------------------

    /// Register a push subscription for a wallet.
    ///
    /// Validates endpoint URL against known push service prefixes and
    /// enforces a per-wallet subscription cap.
    pub async fn subscribe(
        &self,
        wallet: &str,
        endpoint: &str,
        auth_key: &str,
        p256dh_key: &str,
        preferences: Option<serde_json::Value>,
    ) -> Result<PushSubscriptionRecord> {
        // Validate endpoint URL against allowed push service prefixes (SSRF prevention)
        if !ALLOWED_PUSH_ENDPOINT_PREFIXES
            .iter()
            .any(|prefix| endpoint.starts_with(prefix))
        {
            anyhow::bail!("Push endpoint URL is not from a recognized push service");
        }

        let wallet_lower = wallet.to_lowercase();

        // Enforce per-wallet subscription cap
        let existing_count = self
            .db
            .count_push_subscriptions(&wallet_lower)
            .await
            .context("Failed to count existing push subscriptions")?;

        if existing_count >= MAX_SUBSCRIPTIONS_PER_WALLET {
            // Check if this is an upsert (existing endpoint) before rejecting
            let is_upsert = self
                .db
                .push_subscription_exists(&wallet_lower, endpoint)
                .await
                .unwrap_or(false);
            if !is_upsert {
                anyhow::bail!(
                    "Maximum push subscriptions ({MAX_SUBSCRIPTIONS_PER_WALLET}) reached for this wallet"
                );
            }
        }

        let prefs = preferences.unwrap_or_else(default_preferences);

        let row = self
            .db
            .upsert_push_subscription(UpsertPushSubscriptionInput {
                wallet_address: wallet_lower.clone(),
                endpoint: endpoint.to_string(),
                auth_key: auth_key.to_string(),
                p256dh_key: p256dh_key.to_string(),
                preferences: prefs,
            })
            .await
            .context("Failed to upsert push subscription")?;

        self.cache.remove(&wallet_lower);

        info!(wallet = wallet, "Push subscription registered");
        Ok(row)
    }

    /// Update notification preferences for an existing subscription.
    pub async fn update_preferences(
        &self,
        wallet: &str,
        endpoint: &str,
        preferences: serde_json::Value,
    ) -> Result<bool> {
        let wallet_lower = wallet.to_lowercase();

        let updated = self
            .db
            .update_push_preferences(&wallet_lower, endpoint, preferences)
            .await
            .context("Failed to update push preferences")?;

        self.cache.remove(&wallet_lower);

        Ok(updated)
    }

    /// Remove a push subscription by endpoint.
    pub async fn unsubscribe(&self, wallet: &str, endpoint: &str) -> Result<bool> {
        let wallet_lower = wallet.to_lowercase();

        let deleted = self
            .db
            .delete_push_subscription(&wallet_lower, endpoint)
            .await
            .context("Failed to delete push subscription")?;

        self.cache.remove(&wallet_lower);

        Ok(deleted)
    }

    // -- High-level event helpers -------------------------------------------
    // These enqueue a job onto the bounded channel. If the channel is full
    // the job is dropped (try_send). Push is best-effort.

    /// Enqueue a fill push notification for a wallet.
    pub fn send_fill_notification(
        &self,
        wallet: String,
        action: &str,
        size: impl std::fmt::Display,
        symbol: String,
        price: impl std::fmt::Display,
        trade_id: u64,
        realized_pnl: Option<rust_decimal::Decimal>,
    ) {
        let pnl_str = match realized_pnl {
            Some(pnl) if !pnl.is_zero() => format!(" (PnL: ${pnl})"),
            _ => String::new(),
        };
        self.enqueue(PushJob {
            wallet,
            notif_type: NotificationType::Fill,
            payload: PushPayload {
                title: "Order Filled".to_string(),
                body: format!("{action} {size} {symbol} @ ${price}{pnl_str}"),
                tag: format!("fill-{trade_id}"),
                url: None,
            },
        });
    }

    /// Enqueue a liquidation state change push notification.
    pub fn send_liquidation_notification(
        &self,
        wallet: String,
        previous_state: impl std::fmt::Display,
        new_state: impl std::fmt::Display,
        title: &str,
    ) {
        self.enqueue(PushJob {
            wallet,
            notif_type: NotificationType::Liquidation,
            payload: PushPayload {
                title: title.to_string(),
                body: format!("Account status: {previous_state} -> {new_state}"),
                tag: "liquidation".to_string(),
                url: None,
            },
        });
    }

    fn enqueue(&self, job: PushJob) {
        metrics::counter!("ht_push_enqueued_total", "type" => job.notif_type.preference_key())
            .increment(1);
        if self.job_tx.try_send(job).is_err() {
            metrics::counter!("ht_push_dropped_total").increment(1);
            warn!("Push notification queue full, dropping message");
        }
    }
}

// ---------------------------------------------------------------------------
// Background sender task
// ---------------------------------------------------------------------------

/// Load subscriptions for a wallet, using cache when available.
async fn get_subscriptions(
    wallet: &str,
    db: &dyn PushSubscriptionReader,
    cache: &DashMap<String, Vec<PushSubscriptionRecord>>,
) -> Vec<PushSubscriptionRecord> {
    let wallet_lower = wallet.to_lowercase();

    // Fast path: cached
    if let Some(cached) = cache.get(&wallet_lower) {
        return cached.clone();
    }

    // Slow path: DB query
    let subs = match db.get_push_subscriptions(&wallet_lower).await {
        Ok(s) => s,
        Err(e) => {
            warn!(wallet, "Failed to load push subscriptions: {e}");
            return vec![];
        }
    };

    // Evict entire cache if it grows too large
    if cache.len() >= CACHE_MAX_ENTRIES {
        warn!(
            entries = cache.len(),
            "Push subscription cache exceeded max size, clearing"
        );
        cache.clear();
    }

    cache.insert(wallet_lower, subs.clone());
    subs
}

async fn remove_stale(
    id: i64,
    wallet_lower: &str,
    db: &dyn PushSubscriptionWriter,
    cache: &DashMap<String, Vec<PushSubscriptionRecord>>,
) {
    let _ = db.delete_push_subscription_by_id(id).await;
    cache.remove(wallet_lower);
    debug!(id, "Removed stale push subscription");
}

/// Normalize a base64 or base64url string to base64url (no padding).
fn to_base64url(s: &str) -> String {
    s.replace('+', "-")
        .replace('/', "_")
        .trim_end_matches('=')
        .to_string()
}

/// Single background task that drains the push job queue and sends
/// notifications with bounded concurrency.
async fn push_sender_loop(
    mut rx: mpsc::Receiver<PushJob>,
    db: Arc<dyn PushSubscriptionWriter>,
    client: Arc<reqwest::Client>,
    cache: Arc<DashMap<String, Vec<PushSubscriptionRecord>>>,
    vapid_key: Arc<Vec<u8>>,
) {
    let semaphore = Arc::new(tokio::sync::Semaphore::new(MAX_CONCURRENT_SENDS));

    while let Some(job) = rx.recv().await {
        let subs = get_subscriptions(&job.wallet, db.as_ref(), &cache).await;
        if subs.is_empty() {
            continue;
        }

        let json = match serde_json::to_vec(&job.payload) {
            Ok(j) => j,
            Err(e) => {
                warn!("Failed to serialize push payload: {e}");
                continue;
            }
        };

        let wallet_lower = job.wallet.to_lowercase();

        for sub in subs {
            if !is_enabled(&sub.preferences, job.notif_type) {
                continue;
            }

            // Defense-in-depth: skip endpoints that don't match allowed prefixes
            if !ALLOWED_PUSH_ENDPOINT_PREFIXES
                .iter()
                .any(|prefix| sub.endpoint.starts_with(prefix))
            {
                warn!(endpoint = %sub.endpoint, "Skipping push to unrecognized endpoint");
                continue;
            }

            // Acquire a semaphore permit to bound concurrency
            let permit = match semaphore.clone().acquire_owned().await {
                Ok(p) => p,
                Err(_) => break, // Semaphore closed, shutting down
            };

            let client = client.clone();
            let vapid_key = vapid_key.clone();
            let json = json.clone();
            let db = db.clone();
            let cache = cache.clone();
            let wallet_lower = wallet_lower.clone();
            let wallet_for_log = job.wallet.clone();

            tokio::spawn(async move {
                let _permit = permit; // Hold until done

                // Normalize keys to base64url (browser may send standard base64)
                let p256dh = to_base64url(&sub.p256dh_key);
                let auth = to_base64url(&sub.auth_key);
                let info = SubscriptionInfo::new(&sub.endpoint, &p256dh, &auth);

                let partial_builder = match VapidSignatureBuilder::from_pem_no_sub(
                    std::io::Cursor::new(vapid_key.as_ref()),
                ) {
                    Ok(b) => b,
                    Err(e) => {
                        warn!("Failed to create VAPID builder: {e}");
                        return;
                    }
                };

                let sig = match partial_builder.add_sub_info(&info).build() {
                    Ok(sig) => sig,
                    Err(e) => {
                        warn!("Failed to build VAPID signature: {e}");
                        return;
                    }
                };

                let mut builder = WebPushMessageBuilder::new(&info);
                builder.set_payload(ContentEncoding::Aes128Gcm, &json);
                builder.set_vapid_signature(sig);

                let message = match builder.build() {
                    Ok(m) => m,
                    Err(e) => {
                        warn!("Failed to build push message: {e}");
                        return;
                    }
                };

                // Convert web-push message to HTTP request and send via reqwest
                let http_request = web_push::request_builder::build_request::<Vec<u8>>(message);
                let (parts, body) = http_request.into_parts();
                let url = parts.uri.to_string();

                let mut req = client.post(&url);
                for (name, value) in &parts.headers {
                    if let Ok(v) = value.to_str() {
                        req = req.header(name.as_str(), v);
                    }
                }
                req = req.body(body);

                let send_start = std::time::Instant::now();
                match req.send().await {
                    Ok(resp) => {
                        let status = resp.status().as_u16();
                        if (200..300).contains(&(status as usize)) {
                            metrics::counter!("ht_push_sent_total").increment(1);
                            metrics::histogram!("ht_push_send_duration_seconds")
                                .record(send_start.elapsed().as_secs_f64());
                            debug!(wallet = wallet_for_log, endpoint = %sub.endpoint, status, "Push notification sent");
                        } else {
                            metrics::counter!("ht_push_errors_total").increment(1);
                            let body = resp.text().await.unwrap_or_default();
                            warn!(
                                wallet = wallet_for_log,
                                endpoint = %sub.endpoint,
                                status,
                                body = %body,
                                "Push send failed"
                            );
                            if status == 404 || status == 410 {
                                metrics::counter!("ht_push_stale_removed_total").increment(1);
                                remove_stale(sub.id, &wallet_lower, db.as_ref(), &cache).await;
                            }
                        }
                    }
                    Err(e) => {
                        metrics::counter!("ht_push_errors_total").increment(1);
                        warn!(
                            wallet = wallet_for_log,
                            endpoint = %sub.endpoint,
                            error = %e,
                            "Push HTTP request failed"
                        );
                    }
                }
            });
        }
    }

    info!("Push sender loop exited");
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::time::Instant;

    /// Measure the hot path (cache lookup for a wallet with no subscriptions).
    /// This is the path hit on every fill for wallets without push subscriptions.
    /// No hard assertion on timing since CI machines have variable performance.
    #[test]
    fn cache_lookup_no_subscribers_baseline() {
        let cache: DashMap<String, Vec<PushSubscriptionRecord>> = DashMap::new();
        cache.insert("0xdeadbeef".to_string(), vec![]);

        let iterations = 100_000;
        let start = Instant::now();

        for _ in 0..iterations {
            let result = cache.get("0xdeadbeef");
            assert!(result.is_some());
            assert!(result.unwrap().is_empty());
        }

        let elapsed = start.elapsed();
        let per_op_ns = elapsed.as_nanos() / iterations as u128;

        // Log for manual review; no hard threshold to avoid CI flakes.
        eprintln!(
            "push_service cache lookup: {per_op_ns}ns/op ({iterations} iterations in {:?})",
            elapsed
        );
    }

    /// Verify that cache miss + insert path works correctly.
    #[test]
    fn cache_miss_populates_entry() {
        let cache: DashMap<String, Vec<PushSubscriptionRecord>> = DashMap::new();

        // Miss returns None
        assert!(cache.get("0xunknown").is_none());

        // After insert, hit returns the value
        cache.insert("0xunknown".to_string(), vec![]);
        assert!(cache.get("0xunknown").is_some());
    }

    #[test]
    fn is_enabled_defaults_to_true() {
        let prefs = serde_json::json!({});
        assert!(is_enabled(&prefs, NotificationType::Fill));
        assert!(is_enabled(&prefs, NotificationType::Liquidation));
        assert!(is_enabled(&prefs, NotificationType::Settlement));
    }

    #[test]
    fn is_enabled_respects_false() {
        let prefs = serde_json::json!({
            "fills": true,
            "liquidations": false,
            "settlements": true,
        });
        assert!(is_enabled(&prefs, NotificationType::Fill));
        assert!(!is_enabled(&prefs, NotificationType::Liquidation));
        assert!(is_enabled(&prefs, NotificationType::Settlement));
    }

    #[test]
    fn default_preferences_enables_all() {
        let prefs = default_preferences();
        assert!(is_enabled(&prefs, NotificationType::Fill));
        assert!(is_enabled(&prefs, NotificationType::Liquidation));
        assert!(is_enabled(&prefs, NotificationType::Settlement));
    }

    #[test]
    fn notification_type_preference_keys() {
        assert_eq!(NotificationType::Fill.preference_key(), "fills");
        assert_eq!(
            NotificationType::Liquidation.preference_key(),
            "liquidations"
        );
        assert_eq!(NotificationType::Settlement.preference_key(), "settlements");
    }

    #[test]
    fn cache_eviction_on_max_entries() {
        let cache: DashMap<String, Vec<PushSubscriptionRecord>> = DashMap::new();

        // Fill cache to max
        for i in 0..CACHE_MAX_ENTRIES {
            cache.insert(format!("0x{i:040x}"), vec![]);
        }
        assert_eq!(cache.len(), CACHE_MAX_ENTRIES);

        // Simulate the eviction check in get_subscriptions
        if cache.len() >= CACHE_MAX_ENTRIES {
            cache.clear();
        }
        assert_eq!(cache.len(), 0);
    }

    #[test]
    fn is_enabled_rejects_non_bool_values() {
        // Non-bool values should be treated as disabled (reject malformed)
        let prefs = serde_json::json!({"fills": "yes", "liquidations": 1});
        assert!(!is_enabled(&prefs, NotificationType::Fill));
        assert!(!is_enabled(&prefs, NotificationType::Liquidation));
        // Missing key still defaults to true
        assert!(is_enabled(&prefs, NotificationType::Settlement));
    }

    #[test]
    fn is_enabled_handles_null_preferences() {
        let prefs = serde_json::Value::Null;
        // Null preferences should default to all enabled
        assert!(is_enabled(&prefs, NotificationType::Fill));
        assert!(is_enabled(&prefs, NotificationType::Liquidation));
        assert!(is_enabled(&prefs, NotificationType::Settlement));
    }

    #[test]
    fn try_send_drops_when_full() {
        // Verify that a full channel drops messages without blocking
        let (tx, _rx) = mpsc::channel::<PushJob>(1);

        // Fill the channel
        let result1 = tx.try_send(PushJob {
            wallet: "0x1".to_string(),
            notif_type: NotificationType::Fill,
            payload: PushPayload {
                title: "t".to_string(),
                body: "b".to_string(),
                tag: "tag".to_string(),
                url: None,
            },
        });
        assert!(result1.is_ok());

        // Second send should fail (channel full) without blocking
        let result2 = tx.try_send(PushJob {
            wallet: "0x2".to_string(),
            notif_type: NotificationType::Fill,
            payload: PushPayload {
                title: "t".to_string(),
                body: "b".to_string(),
                tag: "tag".to_string(),
                url: None,
            },
        });
        assert!(result2.is_err());
    }

    #[test]
    fn allowed_endpoint_prefixes_accept_known_services() {
        let valid = [
            "https://fcm.googleapis.com/fcm/send/abc123",
            "https://updates.push.services.mozilla.com/wpush/v2/abc",
            "https://wns.windows.com/w/?token=abc",
            "https://web.push.apple.com/abc",
        ];
        for url in &valid {
            assert!(
                ALLOWED_PUSH_ENDPOINT_PREFIXES
                    .iter()
                    .any(|p| url.starts_with(p)),
                "Expected {url} to be allowed"
            );
        }
    }

    #[test]
    fn allowed_endpoint_prefixes_reject_arbitrary_urls() {
        let invalid = [
            "https://evil.com/steal",
            "http://fcm.googleapis.com/http-not-https",
            "https://example.com",
            "file:///etc/passwd",
            "",
        ];
        for url in &invalid {
            assert!(
                !ALLOWED_PUSH_ENDPOINT_PREFIXES
                    .iter()
                    .any(|p| url.starts_with(p)),
                "Expected {url} to be rejected"
            );
        }
    }
}