Skip to main content

max / makenotwork

11.2 KB · 361 lines History Blame Raw
1 //! Page view tracking with daily aggregation.
2 //!
3 //! Each page view UPSERTs into `page_view_daily`, incrementing a counter per
4 //! (target_type, target_id, date). No raw per-request rows; the table stays
5 //! small (365 rows/item/year).
6
7 use chrono::{DateTime, Utc};
8 use sqlx::PgPool;
9 use uuid::Uuid;
10
11 use super::analytics::{TimeRange, format_bucket_label};
12 use super::{ProjectId, UserId};
13 use crate::error::Result;
14
15 /// Background batcher for page-view UPSERTs. Replaces the previous
16 /// `tokio::spawn(record_view(...))` per request pattern, which under any traffic
17 /// spike took a connection per pageview and starved real requests at the pool
18 /// acquire boundary.
19 ///
20 /// `PageViewTx::try_record` is non-blocking (`try_send`); on channel overflow
21 /// the increment is dropped (view counts are already approximate, bot filter,
22 /// no per-user dedupe, so losing a fraction during a burst is acceptable).
23 /// The background drainer flushes a single bulk UPSERT every `FLUSH_INTERVAL`.
24 const CHANNEL_CAPACITY: usize = 4096;
25 const FLUSH_INTERVAL: std::time::Duration = std::time::Duration::from_millis(500);
26
27 type ViewKey = (&'static str, Uuid);
28
29 #[derive(Clone)]
30 pub struct PageViewTx(tokio::sync::mpsc::Sender<ViewKey>);
31
32 impl PageViewTx {
33 pub fn try_record(&self, target_type: &'static str, target_id: Uuid) {
34 let _ = self.0.try_send((target_type, target_id));
35 }
36 }
37
38 /// Spawn the background drainer and return the sender to install on AppState.
39 pub fn spawn_batcher(pool: PgPool) -> PageViewTx {
40 let (tx, mut rx) = tokio::sync::mpsc::channel::<ViewKey>(CHANNEL_CAPACITY);
41 tokio::spawn(async move {
42 let mut pending: std::collections::HashMap<ViewKey, i64> = std::collections::HashMap::new();
43 let mut tick = tokio::time::interval(FLUSH_INTERVAL);
44 tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
45 loop {
46 tokio::select! {
47 Some(key) = rx.recv() => {
48 *pending.entry(key).or_insert(0) += 1;
49 }
50 _ = tick.tick() => {
51 if pending.is_empty() {
52 continue;
53 }
54 let batch = std::mem::take(&mut pending);
55 if let Err(e) = flush_batch(&pool, &batch).await {
56 tracing::warn!(error = ?e, count = batch.len(), "page-view batch flush failed");
57 }
58 }
59 else => break,
60 }
61 }
62 });
63 PageViewTx(tx)
64 }
65
66 async fn flush_batch(pool: &PgPool, batch: &std::collections::HashMap<ViewKey, i64>) -> Result<()> {
67 // Unzip into parallel arrays for UNNEST-style bulk INSERT, single roundtrip
68 // regardless of batch size.
69 let mut types: Vec<&str> = Vec::with_capacity(batch.len());
70 let mut ids: Vec<Uuid> = Vec::with_capacity(batch.len());
71 let mut counts: Vec<i64> = Vec::with_capacity(batch.len());
72 for ((t, id), c) in batch {
73 types.push(t);
74 ids.push(*id);
75 counts.push(*c);
76 }
77 sqlx::query(
78 r"
79 INSERT INTO page_view_daily (target_type, target_id, view_date, view_count)
80 SELECT t, i, CURRENT_DATE, c
81 FROM UNNEST($1::TEXT[], $2::UUID[], $3::BIGINT[]) AS u(t, i, c)
82 ON CONFLICT (target_type, target_id, view_date)
83 DO UPDATE SET view_count = page_view_daily.view_count + EXCLUDED.view_count
84 ",
85 )
86 .bind(&types)
87 .bind(&ids)
88 .bind(&counts)
89 .execute(pool)
90 .await?;
91 Ok(())
92 }
93
94 /// Direct UPSERT path. Kept for backfill / admin tooling; the request path
95 /// goes through `PageViewTx::try_record` to avoid pool pressure under burst.
96 #[allow(dead_code)]
97 pub async fn record_view(pool: &PgPool, target_type: &str, target_id: Uuid) -> Result<()> {
98 sqlx::query(
99 r"
100 INSERT INTO page_view_daily (target_type, target_id, view_date, view_count)
101 VALUES ($1, $2, CURRENT_DATE, 1)
102 ON CONFLICT (target_type, target_id, view_date)
103 DO UPDATE SET view_count = page_view_daily.view_count + 1
104 ",
105 )
106 .bind(target_type)
107 .bind(target_id)
108 .execute(pool)
109 .await?;
110 Ok(())
111 }
112
113 /// A single time bucket in a view timeseries.
114 #[allow(dead_code)]
115 pub struct ViewBucket {
116 pub label: String,
117 pub view_count: i64,
118 }
119
120 /// Fetch time-bucketed view counts for a seller (across all their items and projects).
121 ///
122 /// Optionally scoped to a single project. Uses the same bucketing as revenue charts.
123 #[allow(dead_code)]
124 pub async fn get_view_timeseries(
125 pool: &PgPool,
126 seller_id: UserId,
127 project_id: Option<ProjectId>,
128 range: &TimeRange,
129 ) -> Result<Vec<ViewBucket>> {
130 let bucket = range.bucket_sql();
131 let time_filter = match range.interval_sql() {
132 Some(interval) => format!(" AND pv.view_date >= (CURRENT_DATE - INTERVAL '{interval}')"),
133 None => String::new(),
134 };
135
136 let project_filter = if project_id.is_some() {
137 " AND i.project_id = $2"
138 } else {
139 ""
140 };
141
142 let sql = format!(
143 r"
144 SELECT
145 date_trunc('{bucket}', pv.view_date::TIMESTAMPTZ) AS bucket,
146 COALESCE(SUM(pv.view_count), 0)::BIGINT
147 FROM page_view_daily pv
148 JOIN items i ON pv.target_type = 'item' AND pv.target_id = i.id
149 JOIN projects p ON i.project_id = p.id
150 WHERE p.user_id = $1{project_filter}{time_filter}
151 GROUP BY bucket
152 ORDER BY bucket
153 LIMIT 500
154 "
155 );
156
157 let rows: Vec<(DateTime<Utc>, i64)> = if let Some(pid) = project_id {
158 sqlx::query_as(&sql)
159 .bind(seller_id)
160 .bind(pid)
161 .fetch_all(pool)
162 .await?
163 } else {
164 sqlx::query_as(&sql).bind(seller_id).fetch_all(pool).await?
165 };
166
167 let buckets = rows
168 .into_iter()
169 .map(|(dt, count)| ViewBucket {
170 label: format_bucket_label(&dt, range),
171 view_count: count,
172 })
173 .collect();
174
175 Ok(buckets)
176 }
177
178 /// Period-over-period view comparison for stat cards.
179 ///
180 /// Returns `(current_views, previous_views)`.
181 pub async fn get_view_period_comparison(
182 pool: &PgPool,
183 seller_id: UserId,
184 project_id: Option<ProjectId>,
185 range: &TimeRange,
186 ) -> Result<(i64, i64)> {
187 let Some(interval) = range.interval_sql() else {
188 // All time: no comparison possible, return total views with 0 previous.
189 let total = get_total_views(pool, seller_id, project_id, None).await?;
190 return Ok((total, 0));
191 };
192
193 let project_filter = if project_id.is_some() {
194 " AND i.project_id = $2"
195 } else {
196 ""
197 };
198
199 let sql = format!(
200 r"
201 SELECT
202 COALESCE(SUM(pv.view_count) FILTER (
203 WHERE pv.view_date >= CURRENT_DATE - INTERVAL '{interval}'
204 ), 0)::BIGINT,
205 COALESCE(SUM(pv.view_count) FILTER (
206 WHERE pv.view_date >= CURRENT_DATE - INTERVAL '{interval}' * 2
207 AND pv.view_date < CURRENT_DATE - INTERVAL '{interval}'
208 ), 0)::BIGINT
209 FROM page_view_daily pv
210 JOIN items i ON pv.target_type = 'item' AND pv.target_id = i.id
211 JOIN projects p ON i.project_id = p.id
212 WHERE p.user_id = $1{project_filter}
213 AND pv.view_date >= CURRENT_DATE - INTERVAL '{interval}' * 2
214 "
215 );
216
217 let row: (i64, i64) = if let Some(pid) = project_id {
218 sqlx::query_as(&sql)
219 .bind(seller_id)
220 .bind(pid)
221 .fetch_one(pool)
222 .await?
223 } else {
224 sqlx::query_as(&sql).bind(seller_id).fetch_one(pool).await?
225 };
226
227 Ok(row)
228 }
229
230 /// Total views for a seller, optionally scoped to a project and time range.
231 async fn get_total_views(
232 pool: &PgPool,
233 seller_id: UserId,
234 project_id: Option<ProjectId>,
235 since: Option<DateTime<Utc>>,
236 ) -> Result<i64> {
237 let time_filter = if since.is_some() {
238 " AND pv.view_date >= $2::DATE"
239 } else {
240 ""
241 };
242 let project_filter = if project_id.is_some() {
243 if since.is_some() {
244 " AND i.project_id = $3"
245 } else {
246 " AND i.project_id = $2"
247 }
248 } else {
249 ""
250 };
251
252 let sql = format!(
253 r"
254 SELECT COALESCE(SUM(pv.view_count), 0)::BIGINT
255 FROM page_view_daily pv
256 JOIN items i ON pv.target_type = 'item' AND pv.target_id = i.id
257 JOIN projects p ON i.project_id = p.id
258 WHERE p.user_id = $1{time_filter}{project_filter}
259 "
260 );
261
262 let row: (i64,) = match (since, project_id) {
263 (Some(s), Some(pid)) => {
264 sqlx::query_as(&sql)
265 .bind(seller_id)
266 .bind(s)
267 .bind(pid)
268 .fetch_one(pool)
269 .await?
270 }
271 (Some(s), None) => {
272 sqlx::query_as(&sql)
273 .bind(seller_id)
274 .bind(s)
275 .fetch_one(pool)
276 .await?
277 }
278 (None, Some(pid)) => {
279 sqlx::query_as(&sql)
280 .bind(seller_id)
281 .bind(pid)
282 .fetch_one(pool)
283 .await?
284 }
285 (None, None) => sqlx::query_as(&sql).bind(seller_id).fetch_one(pool).await?,
286 };
287
288 Ok(row.0)
289 }
290
291 /// Per-project view totals for a seller. Used for the cross-project comparison table.
292 pub async fn get_views_by_seller_projects(
293 pool: &PgPool,
294 seller_id: UserId,
295 range: &TimeRange,
296 ) -> Result<Vec<(ProjectId, i64)>> {
297 let time_filter = match range.interval_sql() {
298 Some(interval) => format!(" AND pv.view_date >= (CURRENT_DATE - INTERVAL '{interval}')"),
299 None => String::new(),
300 };
301
302 let sql = format!(
303 r"
304 SELECT p.id, COALESCE(SUM(pv.view_count), 0)::BIGINT
305 FROM projects p
306 LEFT JOIN items i ON i.project_id = p.id
307 LEFT JOIN page_view_daily pv
308 ON pv.target_type = 'item' AND pv.target_id = i.id{time_filter}
309 WHERE p.user_id = $1
310 GROUP BY p.id
311 "
312 );
313
314 let rows: Vec<(ProjectId, i64)> = sqlx::query_as(&sql).bind(seller_id).fetch_all(pool).await?;
315
316 Ok(rows)
317 }
318
319 /// Per-item view totals for a project. Used for the project analytics "top items" list.
320 #[allow(dead_code)]
321 pub async fn get_views_by_project_items(
322 pool: &PgPool,
323 project_id: ProjectId,
324 range: &TimeRange,
325 ) -> Result<Vec<(super::ItemId, i64)>> {
326 let time_filter = match range.interval_sql() {
327 Some(interval) => format!(" AND pv.view_date >= (CURRENT_DATE - INTERVAL '{interval}')"),
328 None => String::new(),
329 };
330
331 let sql = format!(
332 r"
333 SELECT i.id, COALESCE(SUM(pv.view_count), 0)::BIGINT
334 FROM items i
335 LEFT JOIN page_view_daily pv
336 ON pv.target_type = 'item' AND pv.target_id = i.id{time_filter}
337 WHERE i.project_id = $1
338 GROUP BY i.id
339 "
340 );
341
342 let rows: Vec<(super::ItemId, i64)> = sqlx::query_as(&sql)
343 .bind(project_id)
344 .fetch_all(pool)
345 .await?;
346
347 Ok(rows)
348 }
349
350 /// Delete page view rows older than `retain_days`. Called by the daily scheduler.
351 pub async fn prune_old_views(pool: &PgPool, retain_days: i64) -> Result<u64> {
352 let result = sqlx::query(
353 "DELETE FROM page_view_daily WHERE view_date < CURRENT_DATE - $1 * INTERVAL '1 day'",
354 )
355 .bind(retain_days)
356 .execute(pool)
357 .await?;
358
359 Ok(result.rows_affected())
360 }
361