systemprompt_analytics/repository/funnel/
mutations.rs

1//! Funnel and funnel-progress mutations for `FunnelRepository`.
2//!
3//! Creates funnels with their ordered steps, deactivates and deletes them,
4//! and advances or completes per-session progress. All writes go to the write
5//! pool; `record_progress` only moves a session forward, never backward.
6//!
7//! Copyright (c) systemprompt.io — Business Source License 1.1.
8//! See <https://systemprompt.io> for licensing details.
9
10use crate::Result;
11use chrono::Utc;
12use systemprompt_identifiers::{FunnelId, FunnelProgressId, SessionId};
13
14use super::FunnelRepository;
15use crate::models::{CreateFunnelInput, Funnel, FunnelProgress, FunnelStep, FunnelWithSteps};
16
17impl FunnelRepository {
18    pub async fn create_funnel(&self, input: &CreateFunnelInput) -> Result<FunnelWithSteps> {
19        let funnel_id = FunnelId::generate();
20        let now = Utc::now();
21
22        sqlx::query!(
23            r#"
24            INSERT INTO funnels (id, name, description, is_active, created_at, updated_at)
25            VALUES ($1, $2, $3, TRUE, $4, $4)
26            "#,
27            funnel_id.as_str(),
28            input.name,
29            input.description,
30            now
31        )
32        .execute(&*self.write_pool)
33        .await?;
34
35        let mut funnel_ids_arr = Vec::with_capacity(input.steps.len());
36        let mut step_orders = Vec::with_capacity(input.steps.len());
37        let mut names = Vec::with_capacity(input.steps.len());
38        let mut patterns = Vec::with_capacity(input.steps.len());
39        let mut match_types = Vec::with_capacity(input.steps.len());
40        let mut steps = Vec::with_capacity(input.steps.len());
41
42        for (idx, step_input) in input.steps.iter().enumerate() {
43            let step_order = i32::try_from(idx).unwrap_or(0);
44            funnel_ids_arr.push(funnel_id.as_str().to_owned());
45            step_orders.push(step_order);
46            names.push(step_input.name.clone());
47            patterns.push(step_input.match_pattern.clone());
48            match_types.push(step_input.match_type.as_str().to_owned());
49
50            steps.push(FunnelStep {
51                funnel_id: funnel_id.clone(),
52                step_order,
53                name: step_input.name.clone(),
54                match_pattern: step_input.match_pattern.clone(),
55                match_type: step_input.match_type,
56            });
57        }
58
59        if !steps.is_empty() {
60            sqlx::query!(
61                r#"
62                INSERT INTO funnel_steps (funnel_id, step_order, name, match_pattern, match_type)
63                SELECT * FROM UNNEST($1::text[], $2::int4[], $3::text[], $4::text[], $5::text[])
64                "#,
65                &funnel_ids_arr,
66                &step_orders,
67                &names,
68                &patterns,
69                &match_types
70            )
71            .execute(&*self.write_pool)
72            .await?;
73        }
74
75        let funnel = Funnel {
76            id: funnel_id,
77            name: input.name.clone(),
78            description: input.description.clone(),
79            is_active: true,
80            created_at: now,
81            updated_at: now,
82        };
83
84        Ok(FunnelWithSteps { funnel, steps })
85    }
86
87    pub async fn deactivate(&self, id: &FunnelId) -> Result<bool> {
88        let result = sqlx::query!(
89            r#"
90            UPDATE funnels SET is_active = FALSE, updated_at = $2 WHERE id = $1
91            "#,
92            id.as_str(),
93            Utc::now()
94        )
95        .execute(&*self.write_pool)
96        .await?;
97
98        Ok(result.rows_affected() > 0)
99    }
100
101    pub async fn delete(&self, id: &FunnelId) -> Result<bool> {
102        let result = sqlx::query!(r#"DELETE FROM funnels WHERE id = $1"#, id.as_str())
103            .execute(&*self.write_pool)
104            .await?;
105
106        Ok(result.rows_affected() > 0)
107    }
108
109    pub async fn record_progress(
110        &self,
111        funnel_id: &FunnelId,
112        session_id: &SessionId,
113        step: i32,
114    ) -> Result<FunnelProgress> {
115        let now = Utc::now();
116        let step_timestamp = serde_json::json!({
117            "step": step,
118            "timestamp": now.to_rfc3339()
119        });
120
121        if let Some(mut progress) = self.find_progress(funnel_id, session_id).await? {
122            if step > progress.current_step {
123                let mut timestamps = progress
124                    .step_timestamps
125                    .as_array()
126                    .cloned()
127                    .unwrap_or_else(Vec::new);
128                timestamps.push(step_timestamp);
129
130                sqlx::query!(
131                    r#"
132                    UPDATE funnel_progress
133                    SET current_step = $3, step_timestamps = $4, updated_at = $5
134                    WHERE funnel_id = $1 AND session_id = $2
135                    "#,
136                    funnel_id.as_str(),
137                    session_id.as_str(),
138                    step,
139                    serde_json::Value::Array(timestamps.clone()),
140                    now
141                )
142                .execute(&*self.write_pool)
143                .await?;
144
145                progress.current_step = step;
146                progress.step_timestamps = serde_json::Value::Array(timestamps);
147                progress.updated_at = now;
148            }
149            return Ok(progress);
150        }
151
152        let id = FunnelProgressId::generate();
153        let timestamps = serde_json::json!([step_timestamp]);
154
155        sqlx::query!(
156            r#"
157            INSERT INTO funnel_progress (
158                id, funnel_id, session_id, current_step, step_timestamps, created_at, updated_at
159            )
160            VALUES ($1, $2, $3, $4, $5, $6, $6)
161            "#,
162            id.as_str(),
163            funnel_id.as_str(),
164            session_id.as_str(),
165            step,
166            timestamps,
167            now
168        )
169        .execute(&*self.write_pool)
170        .await?;
171
172        Ok(FunnelProgress {
173            id,
174            funnel_id: funnel_id.clone(),
175            session_id: session_id.clone(),
176            current_step: step,
177            completed_at: None,
178            dropped_at_step: None,
179            step_timestamps: timestamps,
180            created_at: now,
181            updated_at: now,
182        })
183    }
184
185    pub async fn mark_completed(
186        &self,
187        funnel_id: &FunnelId,
188        session_id: &SessionId,
189    ) -> Result<bool> {
190        let now = Utc::now();
191        let result = sqlx::query!(
192            r#"
193            UPDATE funnel_progress
194            SET completed_at = $3, updated_at = $3
195            WHERE funnel_id = $1 AND session_id = $2
196            "#,
197            funnel_id.as_str(),
198            session_id.as_str(),
199            now
200        )
201        .execute(&*self.write_pool)
202        .await?;
203
204        Ok(result.rows_affected() > 0)
205    }
206}
systemprompt_analytics/repository/funnel/mutations.rs

systemprompt_analytics/repository/funnel/
mutations.rs
