rovenue/attributes/

dispatcher.rs

use std::sync::Arc;
use std::time::Duration;

use crate::attributes::buffer::AttributeBuffer;
use crate::error::RovenueResult;
use crate::polling::scheduler::PollingScheduler;
use crate::transport::http_client::HttpClient;

/// Most mutations one POST carries.
const BATCH_LIMIT: usize = 200;

pub struct AttributeDispatcher {
    buffer: Arc<AttributeBuffer>,
    http: Arc<HttpClient>,
}

impl AttributeDispatcher {
    pub fn new(buffer: Arc<AttributeBuffer>, http: Arc<HttpClient>) -> Self {
        Self { buffer, http }
    }

    /// List → POST → delete-on-success, oldest first, each mutation under the
    /// identity it was made under (a log out in between changes nothing).
    /// One POST carries the leading run of up to 200 rows that share a
    /// `rovenue_id`, coalesced within that run only; runs are sent in order
    /// until the queue is empty or a POST fails. Returns the number of
    /// distinct keys synced. On a retryable error the run stays queued.
    pub fn flush_once(&self) -> RovenueResult<usize> {
        let mut synced = 0;
        loop {
            let rows = self.buffer.list(BATCH_LIMIT)?;
            let Some(first) = rows.first() else {
                return Ok(synced);
            };
            let sub_id = first.rovenue_id.clone();
            let run: Vec<_> = rows
                .into_iter()
                .take_while(|r| r.rovenue_id == sub_id)
                .collect();
            let ids: Vec<i64> = run.iter().map(|r| r.id).collect();
            // Unattributable (never expected: the identity is minted at
            // configure). Kept, it would block every later mutation.
            if sub_id.is_empty() {
                self.buffer.delete(&ids)?;
                continue;
            }
            // Coalesce in id ASC order: later set of the same key wins.
            let mut map = serde_json::Map::new();
            for r in &run {
                let v = match &r.value {
                    Some(s) => serde_json::Value::String(s.clone()),
                    None => serde_json::Value::Null,
                };
                map.insert(r.key.clone(), v);
            }
            // Post first; only delete if it succeeded (durable). A permanent
            // rejection (4xx other than 429) is the one exception: retaining a
            // run the server will never accept re-sends it every tick and
            // blocks newer mutations forever, so drop it instead. An
            // unresolvable subscriber is not a rejection of the payload: the
            // project owner repairing the merge chain makes the same run
            // deliverable, so it is kept.
            if let Err(err) = self.http.post_attributes(&sub_id, &map) {
                if !err.retryable
                    && err.kind != crate::error::ErrorKind::SubscriberUnresolvable
                    && matches!(err.http_status, Some(s) if (400..500).contains(&s))
                {
                    self.buffer.delete(&ids)?;
                }
                return Err(err);
            }
            self.buffer.delete(&ids)?;
            // Distinct keys synced (coalesced), not raw queue rows drained.
            synced += map.len();
        }
    }

    /// Register the periodic flush on the scheduler (30s, same cadence
    /// as sessions; only fires while foregrounded).
    pub fn start(self: Arc<Self>, scheduler: &PollingScheduler) {
        let me = Arc::clone(&self);
        scheduler.register("attributes", Duration::from_secs(30), move || {
            let _ = me.flush_once();
        });
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::attributes::buffer::AttributeBuffer;
    use crate::cache::store::CacheStore;
    use std::sync::Arc;
    use std::time::Duration;

    // An HttpClient pointed at an unroutable address with a single, short
    // attempt — any actual request fails fast with a network error. Used by
    // the no-op and network-error cases (the no-op case never sends).
    fn http_unreachable() -> Arc<HttpClient> {
        Arc::new(
            HttpClient::new("http://127.0.0.1:1".to_string(), "pk_test_abc".into())
                .with_max_attempts(1)
                .with_min_backoff(Duration::from_millis(1))
                .with_request_timeout(Duration::from_millis(200)),
        )
    }

    #[test]
    fn flush_drops_unattributable_rows_without_sending() {
        let store = Arc::new(CacheStore::open_in_memory().unwrap());
        let buf = Arc::new(AttributeBuffer::new(Arc::clone(&store)));
        buf.set("", "$email", Some("a@b.com")).unwrap();
        // http would fail if called; an empty stamp is never sent.
        let dispatcher = AttributeDispatcher::new(Arc::clone(&buf), http_unreachable());
        assert_eq!(dispatcher.flush_once().unwrap(), 0);
        assert_eq!(buf.list(100).unwrap().len(), 0);
    }

    #[test]
    fn flush_deletes_only_on_success() {
        let mut server = mockito::Server::new();
        let m = server
            .mock("POST", "/v1/me/attributes")
            .match_header("x-rovenue-app-user-id", "rov_x")
            .match_body(mockito::Matcher::JsonString(
                r#"{"attributes":{"$email":"a@b.com","country":null}}"#.into(),
            ))
            .with_status(200)
            .with_body(r#"{"data":{"ok":true}}"#)
            .create();

        let store = Arc::new(CacheStore::open_in_memory().unwrap());
        let buf = Arc::new(AttributeBuffer::new(Arc::clone(&store)));
        buf.set("rov_x", "$email", Some("a@b.com")).unwrap();
        buf.set("rov_x", "country", None).unwrap();
        let http = Arc::new(
            HttpClient::new(server.url(), "pk_test_abc".into())
                .with_max_attempts(1)
                .with_request_timeout(Duration::from_millis(500)),
        );
        let dispatcher = AttributeDispatcher::new(Arc::clone(&buf), http);
        assert_eq!(dispatcher.flush_once().unwrap(), 2);
        assert_eq!(buf.list(100).unwrap().len(), 0);
        m.assert();
    }

    /// Same poison-pill guard as the session dispatcher: a queue the server
    /// permanently rejects (4xx other than 429) must be dropped, or every
    /// future flush re-sends it and newer mutations never sync.
    #[test]
    fn flush_discards_queue_on_permanent_4xx() {
        let mut server = mockito::Server::new();
        let _m = server
            .mock("POST", "/v1/me/attributes")
            .with_status(400)
            .with_body(r#"{"error":{"code":"VALIDATION_ERROR"}}"#)
            .create();

        let store = Arc::new(CacheStore::open_in_memory().unwrap());
        let buf = Arc::new(AttributeBuffer::new(Arc::clone(&store)));
        buf.set("rov_x", "$email", Some("a@b.com")).unwrap();
        let http = Arc::new(
            HttpClient::new(server.url(), "pk_test_abc".into())
                .with_max_attempts(1)
                .with_request_timeout(Duration::from_millis(500)),
        );
        let dispatcher = AttributeDispatcher::new(Arc::clone(&buf), http);
        assert!(dispatcher.flush_once().is_err(), "a 400 must still surface");
        assert_eq!(
            buf.list(100).unwrap().len(),
            0,
            "a 4xx-rejected queue must be dropped, not retried forever"
        );
    }

    #[test]
    fn flush_keeps_queue_on_network_error() {
        let store = Arc::new(CacheStore::open_in_memory().unwrap());
        let buf = Arc::new(AttributeBuffer::new(Arc::clone(&store)));
        buf.set("rov_x", "$email", Some("a@b.com")).unwrap();
        let dispatcher = AttributeDispatcher::new(Arc::clone(&buf), http_unreachable());
        assert!(dispatcher.flush_once().is_err());
        // NOT deleted — durable retry
        assert_eq!(buf.list(100).unwrap().len(), 1);
    }
}