Skip to main content

max / makenotwork

6.3 KB · 193 lines History Blame Raw
1 //! SyncKit per-user devices: upsert a paired device, list a user's devices,
2 //! verify ownership, advance its sync cursor (with a last-seen touch), and
3 //! delete it.
4
5 use sqlx::PgPool;
6
7 use crate::db::enums::SyncPlatform;
8 use crate::db::models::DbSyncDevice;
9 use crate::db::{SyncAppId, SyncDeviceId, UserId};
10 use crate::error::Result;
11
12 // ── Sync Devices ──
13
14 /// Register or update a sync device.
15 ///
16 /// Upserts on the `(app_id, user_id, device_name)` unique constraint.
17 /// On conflict (same device re-registering), updates the platform string
18 /// and bumps `last_seen_at` so stale-device detection stays accurate.
19 ///
20 /// `client_version` is the SDK version off the request's User-Agent. `None`
21 /// means the request carried no recognisable one, and it leaves any previously
22 /// recorded version in place rather than blanking it: a client that stops
23 /// sending the header has not downgraded to unknown, it is just quiet.
24 #[tracing::instrument(skip_all)]
25 pub async fn upsert_sync_device(
26 pool: &PgPool,
27 app_id: SyncAppId,
28 user_id: UserId,
29 device_name: &str,
30 platform: SyncPlatform,
31 client_version: Option<&str>,
32 ) -> Result<DbSyncDevice> {
33 let device = sqlx::query_as::<_, DbSyncDevice>(
34 r"
35 INSERT INTO sync_devices (app_id, user_id, device_name, platform, client_version)
36 VALUES ($1, $2, $3, $4, $5)
37 ON CONFLICT (app_id, user_id, device_name)
38 DO UPDATE SET
39 platform = EXCLUDED.platform,
40 last_seen_at = NOW(),
41 client_version = COALESCE(EXCLUDED.client_version, sync_devices.client_version)
42 RETURNING *
43 ",
44 )
45 .bind(app_id)
46 .bind(user_id)
47 .bind(device_name)
48 .bind(platform)
49 .bind(client_version)
50 .fetch_one(pool)
51 .await?;
52
53 Ok(device)
54 }
55
56 /// List all devices for a user within an app.
57 #[tracing::instrument(skip_all)]
58 pub async fn get_sync_devices(
59 pool: &PgPool,
60 app_id: SyncAppId,
61 user_id: UserId,
62 ) -> Result<Vec<DbSyncDevice>> {
63 let devices = sqlx::query_as::<_, DbSyncDevice>(
64 "SELECT * FROM sync_devices WHERE app_id = $1 AND user_id = $2 ORDER BY last_seen_at DESC LIMIT 100",
65 )
66 .bind(app_id)
67 .bind(user_id)
68 .fetch_all(pool)
69 .await?;
70
71 Ok(devices)
72 }
73
74 /// Does this device belong to this `(app, user)`? Indexed point lookup on the
75 /// `sync_devices` PK + owner columns, the hot push/pull paths use this instead
76 /// of fetching every device and linear-scanning, so device verification stays
77 /// O(1) as a user's device count grows.
78 #[tracing::instrument(skip_all)]
79 pub async fn sync_device_belongs(
80 pool: &PgPool,
81 device_id: SyncDeviceId,
82 app_id: SyncAppId,
83 user_id: UserId,
84 ) -> Result<bool> {
85 let exists: bool = sqlx::query_scalar(
86 "SELECT EXISTS(SELECT 1 FROM sync_devices WHERE id = $1 AND app_id = $2 AND user_id = $3)",
87 )
88 .bind(device_id)
89 .bind(app_id)
90 .bind(user_id)
91 .fetch_one(pool)
92 .await?;
93 Ok(exists)
94 }
95
96 /// Mark a device seen and advance its pull cursor in one statement, the pull
97 /// hot path used to issue a separate touch and a separate cursor update; this
98 /// folds both into a single UPDATE. `GREATEST` keeps the cursor monotonic.
99 ///
100 /// Also refreshes `client_version` from the request that triggered the pull, so
101 /// a device that upgrades reports its new version on its next sync instead of
102 /// waiting for a re-registration that may never come. `None` leaves the stored
103 /// value alone (see [`upsert_sync_device`]).
104 #[tracing::instrument(skip_all)]
105 pub async fn touch_and_advance_cursor(
106 pool: &PgPool,
107 device_id: SyncDeviceId,
108 new_cursor: i64,
109 client_version: Option<&str>,
110 ) -> Result<()> {
111 sqlx::query(
112 "UPDATE sync_devices
113 SET last_seen_at = NOW(),
114 last_pulled_seq = GREATEST(last_pulled_seq, $2),
115 client_version = COALESCE($3, client_version)
116 WHERE id = $1",
117 )
118 .bind(device_id)
119 .bind(new_cursor)
120 .bind(client_version)
121 .execute(pool)
122 .await?;
123 Ok(())
124 }
125
126 /// Delete a device by ID (only if owned by user within app).
127 #[tracing::instrument(skip_all)]
128 pub async fn delete_sync_device(
129 pool: &PgPool,
130 device_id: SyncDeviceId,
131 app_id: SyncAppId,
132 user_id: UserId,
133 ) -> Result<bool> {
134 let result =
135 sqlx::query("DELETE FROM sync_devices WHERE id = $1 AND app_id = $2 AND user_id = $3")
136 .bind(device_id)
137 .bind(app_id)
138 .bind(user_id)
139 .execute(pool)
140 .await?;
141
142 Ok(result.rows_affected() > 0)
143 }
144
145 /// One row of the field-version readout: how many devices last synced on a
146 /// given SDK version, and when the most recent of them was seen.
147 #[derive(Debug, Clone, sqlx::FromRow, serde::Serialize)]
148 pub struct ClientVersionCount {
149 /// SDK version, or `None` for devices that never reported one.
150 pub client_version: Option<String>,
151 /// Devices last seen on that version within the window.
152 pub devices: i64,
153 /// Most recent sync from any device on that version.
154 pub last_seen_at: chrono::DateTime<chrono::Utc>,
155 }
156
157 /// Distribution of SyncKit versions across devices that synced within the last
158 /// `window_days`. Answers "which SyncKit is in the field", which is a different
159 /// question from "which SyncKit did we publish".
160 ///
161 /// The window matters: without it, a laptop that synced once in 2025 and was
162 /// never opened again would keep an ancient version in the readout forever and
163 /// make the fleet look more stale than it is.
164 #[tracing::instrument(skip_all)]
165 pub async fn client_version_distribution(
166 pool: &PgPool,
167 window_days: i32,
168 ) -> Result<Vec<ClientVersionCount>> {
169 let rows = sqlx::query_as::<_, ClientVersionCount>(
170 "SELECT client_version, COUNT(*) AS devices, MAX(last_seen_at) AS last_seen_at
171 FROM sync_devices
172 WHERE last_seen_at > NOW() - make_interval(days => $1)
173 GROUP BY client_version
174 ORDER BY devices DESC, client_version DESC NULLS LAST",
175 )
176 .bind(window_days)
177 .fetch_all(pool)
178 .await?;
179
180 Ok(rows)
181 }
182
183 /// Touch last_seen_at for a device.
184 #[tracing::instrument(skip_all)]
185 pub async fn touch_sync_device(pool: &PgPool, device_id: SyncDeviceId) -> Result<()> {
186 sqlx::query("UPDATE sync_devices SET last_seen_at = NOW() WHERE id = $1")
187 .bind(device_id)
188 .execute(pool)
189 .await?;
190
191 Ok(())
192 }
193