rovenue/cache/
funnel.rs

1use crate::cache::CacheStore;
2use crate::error::RovenueResult;
3
4/// Persists the per-install id and the once-per-install claim state.
5pub struct FunnelRepo<'a> {
6    store: &'a CacheStore,
7}
8
9impl<'a> FunnelRepo<'a> {
10    pub fn new(store: &'a CacheStore) -> Self {
11        Self { store }
12    }
13
14    /// Returns the persisted `install_id`, generating + storing it on first call.
15    pub fn get_or_create_install_id(&self, now_ms: u64) -> RovenueResult<String> {
16        self.store.with_conn(|c| {
17            let existing: Option<String> = c
18                .query_row(
19                    "SELECT install_id FROM funnel_install WHERE id = 1",
20                    [],
21                    |r| r.get(0),
22                )
23                .ok();
24            if let Some(id) = existing {
25                return Ok(id);
26            }
27            let id = format!("inst_{}", cuid2::create_id());
28            c.execute(
29                "INSERT INTO funnel_install (id, install_id, created_at_ms) VALUES (1, ?1, ?2)",
30                rusqlite::params![id, now_ms as i64],
31            )?;
32            Ok(id)
33        })
34    }
35
36    /// Upserts the claim state for an install (`pending`/`claimed`/`failed`).
37    pub fn set_claim_state(
38        &self,
39        install_id: &str,
40        state: &str,
41        subscriber_id: Option<&str>,
42        now_ms: u64,
43    ) -> RovenueResult<()> {
44        self.store.with_conn(|c| {
45            c.execute(
46                "INSERT INTO funnel_claim_state \
47                   (install_id, state, subscriber_id, claimed_at_ms, created_at_ms) \
48                 VALUES (?1, ?2, ?3, ?4, ?5) \
49                 ON CONFLICT(install_id) DO UPDATE SET \
50                   state = excluded.state, \
51                   subscriber_id = excluded.subscriber_id, \
52                   claimed_at_ms = excluded.claimed_at_ms",
53                rusqlite::params![
54                    install_id,
55                    state,
56                    subscriber_id,
57                    if state == "claimed" {
58                        Some(now_ms as i64)
59                    } else {
60                        None::<i64>
61                    },
62                    now_ms as i64
63                ],
64            )?;
65            Ok(())
66        })
67    }
68
69    /// Current claim state for an install, or `None` if never attempted.
70    pub fn claim_state(&self, install_id: &str) -> RovenueResult<Option<String>> {
71        self.store.with_conn(|c| {
72            let s: Option<String> = c
73                .query_row(
74                    "SELECT state FROM funnel_claim_state WHERE install_id = ?1",
75                    [install_id],
76                    |r| r.get(0),
77                )
78                .ok();
79            Ok(s)
80        })
81    }
82}
83
84#[cfg(test)]
85mod tests {
86    use super::*;
87
88    fn store() -> CacheStore {
89        CacheStore::open_in_memory().expect("open in-memory store")
90    }
91
92    #[test]
93    fn install_id_is_stable_across_calls() {
94        let s = store();
95        let repo = FunnelRepo::new(&s);
96        let a = repo.get_or_create_install_id(1000).unwrap();
97        let b = repo.get_or_create_install_id(2000).unwrap();
98        assert!(a.starts_with("inst_"));
99        assert_eq!(a, b, "install_id must persist, not regenerate");
100    }
101
102    #[test]
103    fn claim_state_roundtrips() {
104        let s = store();
105        let repo = FunnelRepo::new(&s);
106        assert_eq!(repo.claim_state("inst_x").unwrap(), None);
107        repo.set_claim_state("inst_x", "claimed", Some("sub_1"), 5000)
108            .unwrap();
109        assert_eq!(repo.claim_state("inst_x").unwrap(), Some("claimed".into()));
110    }
111
112    #[test]
113    fn claimed_at_ms_only_set_when_claimed() {
114        let s = store();
115        let repo = FunnelRepo::new(&s);
116        repo.set_claim_state("inst_p", "pending", None, 1000)
117            .unwrap();
118        repo.set_claim_state("inst_c", "claimed", Some("sub_1"), 2000)
119            .unwrap();
120        let read = |iid: &str| -> Option<i64> {
121            s.with_conn(|c| {
122                c.query_row(
123                    "SELECT claimed_at_ms FROM funnel_claim_state WHERE install_id = ?1",
124                    [iid],
125                    |r| r.get::<_, Option<i64>>(0),
126                )
127            })
128            .unwrap()
129        };
130        assert_eq!(read("inst_p"), None, "pending must not stamp claimed_at_ms");
131        assert_eq!(read("inst_c"), Some(2000));
132    }
133
134    #[test]
135    fn claimed_at_ms_clears_on_conflict_to_pending() {
136        let s = store();
137        let repo = FunnelRepo::new(&s);
138        repo.set_claim_state("inst_flip", "claimed", Some("sub_1"), 2000)
139            .unwrap();
140        repo.set_claim_state("inst_flip", "pending", None, 3000)
141            .unwrap();
142        let read = |iid: &str| -> Option<i64> {
143            s.with_conn(|c| {
144                c.query_row(
145                    "SELECT claimed_at_ms FROM funnel_claim_state WHERE install_id = ?1",
146                    [iid],
147                    |r| r.get::<_, Option<i64>>(0),
148                )
149            })
150            .unwrap()
151        };
152        assert_eq!(
153            read("inst_flip"),
154            None,
155            "claimed→pending flip must clear claimed_at_ms"
156        );
157    }
158}
rovenue/cache/funnel.rs

rovenue/cache/
funnel.rs
