1use std::collections::HashSet;
2use std::path::Path;
3use std::time::Instant;
4
5use anyhow::{Context, Result, bail};
6use sqlx::{QueryBuilder, Sqlite, SqliteConnection};
7
8use super::apply::apply_remote_change;
9use super::wire::{ChangeRow, ChangeWire, PushAck, SyncRequest, SyncResponse};
10use crate::change_log::op_type;
11use crate::db::{Database, begin_immediate, get_meta, set_meta};
12
13#[derive(Debug, Clone, Default, PartialEq, Eq)]
14pub struct SyncPersistenceStatus {
15 pub pinned_server: Option<String>,
16 pub pending_changes: i64,
17 pub conflicts: i64,
18 pub sync_cursor: Option<String>,
19 pub local_sequence: Option<String>,
20 pub last_attempt: Option<String>,
21 pub last_success: Option<String>,
22 pub last_error: Option<String>,
23 pub last_pushed: Option<String>,
24 pub last_pulled: Option<String>,
25 pub last_cursor: Option<String>,
26}
27
28#[derive(Debug, Clone)]
29pub struct ClientSyncPage {
30 pub request: SyncRequest,
31 pub pending: usize,
32}
33
34#[derive(Debug)]
35pub struct ApplySyncPage {
36 pub request: SyncRequest,
37 pub response: SyncResponse,
38 pub attempted_at: String,
39 pub previous_pushed: i64,
40 pub previous_pulled: usize,
41}
42
43#[derive(Debug)]
44pub struct ServerSyncPage {
45 pub request: SyncRequest,
46}
47
48#[derive(Debug)]
49pub struct ServerSyncResult {
50 pub accepted_count: i64,
51 pub push_acks: Vec<PushAck>,
52 pub changes: Vec<ChangeWire>,
53 pub has_more: bool,
54 pub assign_ms: u128,
55 pub pull_query_ms: u128,
56}
57
58impl Database {
59 pub async fn sync_persistence_status(&self) -> Result<SyncPersistenceStatus> {
60 let mut conn = self.acquire_writer().await?;
61 let pending_changes =
62 sqlx::query_scalar("SELECT count(*) FROM changes WHERE server_seq IS NULL")
63 .fetch_one(&mut *conn)
64 .await?;
65 let conflicts = sqlx::query_scalar("SELECT count(*) FROM conflicts WHERE resolved = 0")
66 .fetch_one(&mut *conn)
67 .await?;
68 Ok(SyncPersistenceStatus {
69 pinned_server: get_meta(&mut conn, "sync_server_url").await?,
70 pending_changes,
71 conflicts,
72 sync_cursor: get_meta(&mut conn, "sync_cursor").await?,
73 local_sequence: get_meta(&mut conn, "local_seq").await?,
74 last_attempt: get_meta(&mut conn, "sync_last_attempt_at").await?,
75 last_success: get_meta(&mut conn, "sync_last_success_at").await?,
76 last_error: get_meta(&mut conn, "sync_last_error").await?,
77 last_pushed: get_meta(&mut conn, "sync_last_pushed").await?,
78 last_pulled: get_meta(&mut conn, "sync_last_pulled").await?,
79 last_cursor: get_meta(&mut conn, "sync_last_cursor").await?,
80 })
81 }
82
83 pub async fn begin_sync_attempt(&self, attempted_at: String) -> Result<()> {
84 let mut conn = self.acquire_writer().await?;
85 set_meta(&mut conn, "sync_last_attempt_at", &attempted_at).await
86 }
87
88 pub async fn record_sync_error(&self, error: String) -> Result<()> {
89 let mut conn = self.acquire_writer().await?;
90 set_meta(&mut conn, "sync_last_error", &error).await
91 }
92
93 pub(super) async fn pending_sync_change_count(&self) -> Result<i64> {
94 let mut conn = self.acquire_reader().await?;
95 Ok(
96 sqlx::query_scalar("SELECT COUNT(*) FROM changes WHERE server_seq IS NULL")
97 .fetch_one(&mut *conn)
98 .await?,
99 )
100 }
101
102 pub(super) async fn pending_blob_contracts(
103 &self,
104 ) -> Result<Vec<super::wire::BlobUploadContract>> {
105 let mut conn = self.acquire_reader().await?;
106 let changes = load_unsynced_changes(&mut conn, i64::MAX as usize).await?;
107 super::blob::unique_blob_contracts(&changes)
108 }
109
110 pub async fn prepare_client_sync_page(
111 &self,
112 server: String,
113 push_limit: usize,
114 pull_limit: u32,
115 ) -> Result<ClientSyncPage> {
116 let mut conn = self.acquire_writer().await?;
117 validate_sync_server(&mut conn, &server).await?;
118 let client_id = get_meta(&mut conn, "client_id")
119 .await?
120 .context("missing client id")?;
121 let after = sync_cursor(&mut conn).await?;
122 let changes = load_unsynced_changes(&mut conn, push_limit).await?;
123 let pending = changes.len();
124 Ok(ClientSyncPage {
125 request: SyncRequest {
126 protocol_version: Some(super::wire::SYNC_PROTOCOL_VERSION),
127 client_id,
128 after,
129 pull_limit: Some(pull_limit),
130 changes,
131 },
132 pending,
133 })
134 }
135
136 pub async fn apply_client_sync_page(&self, page: ApplySyncPage) -> Result<usize> {
137 let envelope = super::wire::validate_sync_request_envelope(&page.request)?;
138 let request_change_ids = page
139 .request
140 .changes
141 .iter()
142 .map(|change| change.change_id.clone())
143 .collect::<Vec<_>>();
144 super::wire::validate_sync_response_for_request(
145 envelope.after,
146 envelope.pull_limit,
147 &request_change_ids,
148 &page.response,
149 )?;
150 let mut conn = self.acquire_writer().await?;
151 apply_sync_response(&mut conn, page).await
152 }
153
154 pub async fn persist_server_sync_page(&self, page: ServerSyncPage) -> Result<ServerSyncResult> {
155 self.persist_server_sync_page_inner(page, None).await
156 }
157
158 pub async fn persist_server_sync_page_with_blobs(
159 &self,
160 page: ServerSyncPage,
161 blob_dir: &Path,
162 ) -> Result<ServerSyncResult> {
163 self.persist_server_sync_page_inner(page, Some(blob_dir))
164 .await
165 }
166
167 async fn persist_server_sync_page_inner(
168 &self,
169 page: ServerSyncPage,
170 blob_dir: Option<&Path>,
171 ) -> Result<ServerSyncResult> {
172 let envelope = super::wire::validate_sync_request_envelope(&page.request)?;
173 for change in &page.request.changes {
174 super::wire::validate_pushed_change(change)?;
175 }
176 let mut conn = self.acquire_writer().await?;
177 let assign_started = Instant::now();
178 let (accepted_count, push_acks) =
179 assign_server_sequences(&mut conn, page.request.changes, blob_dir).await?;
180 let assign_ms = assign_started.elapsed().as_millis();
181 let pull_query_started = Instant::now();
182 let (changes, has_more) =
183 load_server_changes_after(&mut conn, envelope.after, envelope.pull_limit).await?;
184 let pull_query_ms = pull_query_started.elapsed().as_millis();
185 Ok(ServerSyncResult {
186 accepted_count,
187 push_acks,
188 changes,
189 has_more,
190 assign_ms,
191 pull_query_ms,
192 })
193 }
194}
195
196async fn sync_cursor(conn: &mut SqliteConnection) -> Result<i64> {
197 Ok(get_meta(conn, "sync_cursor")
198 .await?
199 .unwrap_or_else(|| "0".to_string())
200 .parse::<i64>()?)
201}
202
203async fn validate_sync_server(conn: &mut SqliteConnection, server: &str) -> Result<()> {
204 let normalized = server.trim_end_matches('/');
205 if let Some(existing) = get_meta(conn, "sync_server_url").await? {
206 if existing != normalized {
207 bail!(
208 "error sync-server-changed existing={} requested={} hint=\"use a fresh database for a different sync server\"",
209 existing,
210 normalized
211 );
212 }
213 } else {
214 set_meta(conn, "sync_server_url", normalized).await?;
215 }
216 Ok(())
217}
218
219async fn load_unsynced_changes(
220 conn: &mut SqliteConnection,
221 limit: usize,
222) -> Result<Vec<ChangeWire>> {
223 let limit = limit as i64;
224 let rows = sqlx::query_as!(
225 ChangeRow,
226 r#"SELECT change_id AS "change_id!: String", client_id AS "client_id!: String",
227 local_seq AS "local_seq!: i64", entity_type AS "entity_type!: String",
228 entity_id AS "entity_id!: String", field, op_type AS "op_type!: String",
229 payload AS "payload!: String", base_version, created_at AS "created_at!: String",
230 server_seq
231 FROM changes WHERE server_seq IS NULL ORDER BY local_seq, created_at LIMIT ?"#,
232 limit,
233 )
234 .fetch_all(&mut *conn)
235 .await?;
236 Ok(rows.into_iter().map(ChangeRow::into_wire).collect())
237}
238
239async fn apply_sync_response(conn: &mut SqliteConnection, page: ApplySyncPage) -> Result<usize> {
240 let mut applied = 0;
241 let mut tx = begin_immediate(conn).await?;
242 let current_cursor = sync_cursor(&mut tx).await?;
243 if current_cursor != page.request.after {
244 bail!(
245 "error stale-sync-page expected_cursor={} request_cursor={}",
246 current_cursor,
247 page.request.after
248 );
249 }
250 update_change_server_seqs_if_missing(&mut tx, &page.response.push_acks).await?;
251 let existing_change_ids = load_existing_change_ids(&mut tx, &page.response.changes).await?;
252 let mut affected_series = HashSet::new();
253 for change in &page.response.changes {
254 if existing_change_ids.contains(change.change_id.as_str()) {
255 verify_existing_change(&mut tx, change).await?;
256 update_change_server_seq(&mut tx, &change.change_id, change.server_seq).await?;
257 continue;
258 }
259 apply_remote_change(&mut tx, change).await?;
260 if change.entity_type == "recurrence_series" {
261 let workspace_id = change
262 .payload
263 .get("workspace_id")
264 .and_then(serde_json::Value::as_str)
265 .context("recurrence change missing workspace_id")?;
266 affected_series.insert((workspace_id.to_string(), change.entity_id.clone()));
267 } else if let Some(series_id) = change
268 .payload
269 .get("series_id")
270 .and_then(serde_json::Value::as_str)
271 {
272 let workspace_id = change
273 .payload
274 .get("workspace_id")
275 .and_then(serde_json::Value::as_str)
276 .context("recurrence task change missing workspace_id")?;
277 affected_series.insert((workspace_id.to_string(), series_id.to_string()));
278 }
279 insert_wire_change(&mut tx, change).await?;
280 applied += 1;
281 }
282 for (workspace_id, series_id) in affected_series {
283 let workspace_id: crate::ids::WorkspaceId = workspace_id.parse()?;
284 let series_id: crate::recurrence::RecurrenceSeriesId = series_id.parse()?;
285 let workspace = crate::workspaces::workspace_for_id(&mut tx, &workspace_id).await?;
286 let at =
287 chrono::DateTime::parse_from_rfc3339(&page.attempted_at)?.with_timezone(&chrono::Utc);
288 crate::operations::recurrence::reconcile_recurrence_series_in_transaction(
289 &mut tx, &workspace, &series_id, at,
290 )
291 .await?;
292 }
293 let pushed = page.previous_pushed + page.response.push_acks.len() as i64;
294 let pulled = page.previous_pulled + applied;
295 set_meta(&mut tx, "sync_cursor", &page.response.cursor.to_string()).await?;
296 set_meta(&mut tx, "sync_last_success_at", &page.attempted_at).await?;
297 set_meta(&mut tx, "sync_last_error", "").await?;
298 set_meta(&mut tx, "sync_last_pushed", &pushed.to_string()).await?;
299 set_meta(&mut tx, "sync_last_pulled", &pulled.to_string()).await?;
300 set_meta(
301 &mut tx,
302 "sync_last_cursor",
303 &page.response.cursor.to_string(),
304 )
305 .await?;
306 tx.commit().await?;
307 Ok(applied)
308}
309
310async fn load_existing_change_ids(
311 conn: &mut SqliteConnection,
312 changes: &[ChangeWire],
313) -> Result<HashSet<String>> {
314 if changes.is_empty() {
315 return Ok(HashSet::new());
316 }
317 let mut query_builder =
318 QueryBuilder::<Sqlite>::new("SELECT change_id FROM changes WHERE change_id IN (");
319 let mut separated = query_builder.separated(", ");
320 for change in changes {
321 separated.push_bind(&change.change_id);
322 }
323 separated.push_unseparated(")");
324 Ok(query_builder
325 .build_query_scalar::<String>()
326 .fetch_all(&mut *conn)
327 .await?
328 .into_iter()
329 .collect())
330}
331
332async fn verify_existing_change(conn: &mut SqliteConnection, incoming: &ChangeWire) -> Result<()> {
333 let row = sqlx::query(
334 "SELECT entity_type, entity_id, field, op_type, payload, base_version, created_at
335 FROM changes WHERE change_id = ?",
336 )
337 .bind(&incoming.change_id)
338 .fetch_one(&mut *conn)
339 .await?;
340 use sqlx::Row;
341 let stored_payload: String = row.try_get("payload")?;
342 let stored_payload: serde_json::Value = serde_json::from_str(&stored_payload)?;
343 let equal = row.try_get::<String, _>("entity_type")? == incoming.entity_type
344 && row.try_get::<String, _>("entity_id")? == incoming.entity_id
345 && row.try_get::<Option<String>, _>("field")? == incoming.field
346 && row.try_get::<String, _>("op_type")? == incoming.op_type
347 && stored_payload == incoming.payload
348 && row.try_get::<Option<String>, _>("base_version")? == incoming.base_version
349 && row.try_get::<String, _>("created_at")? == incoming.created_at;
350 if !equal {
351 bail!(
352 "error sync-change-id-payload-mismatch change_id={}",
353 incoming.change_id
354 );
355 }
356 Ok(())
357}
358
359async fn update_change_server_seq(
360 conn: &mut SqliteConnection,
361 change_id: &str,
362 server_seq: Option<i64>,
363) -> Result<()> {
364 if let Some(server_seq) = server_seq {
365 sqlx::query!(
366 "UPDATE changes SET server_seq = ? WHERE change_id = ? AND server_seq IS NULL",
367 server_seq,
368 change_id,
369 )
370 .execute(&mut *conn)
371 .await?;
372 }
373 Ok(())
374}
375
376async fn update_change_server_seqs_if_missing(
377 conn: &mut SqliteConnection,
378 push_acks: &[PushAck],
379) -> Result<()> {
380 if push_acks.is_empty() {
381 return Ok(());
382 }
383 let mut query_builder = QueryBuilder::<Sqlite>::new("WITH updates(change_id, server_seq) AS (");
384 query_builder.push_values(push_acks, |mut row, ack| {
385 row.push_bind(&ack.change_id).push_bind(ack.server_seq);
386 });
387 query_builder.push(
388 ") UPDATE changes
389 SET server_seq = (
390 SELECT updates.server_seq
391 FROM updates
392 WHERE updates.change_id = changes.change_id
393 )
394 WHERE server_seq IS NULL
395 AND change_id IN (SELECT change_id FROM updates)",
396 );
397 query_builder.build().execute(&mut *conn).await?;
398 Ok(())
399}
400
401async fn insert_wire_change(conn: &mut SqliteConnection, change: &ChangeWire) -> Result<()> {
402 let payload = change.payload.to_string();
403 sqlx::query!(
404 "INSERT OR IGNORE INTO changes(change_id, client_id, local_seq, entity_type, entity_id, field,
405 op_type, payload, base_version, created_at, server_seq)
406 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
407 change.change_id,
408 change.client_id,
409 change.local_seq,
410 change.entity_type,
411 change.entity_id,
412 change.field,
413 change.op_type,
414 payload,
415 change.base_version,
416 change.created_at,
417 change.server_seq,
418 )
419 .execute(&mut *conn)
420 .await?;
421 Ok(())
422}
423
424async fn assign_server_sequences(
425 conn: &mut SqliteConnection,
426 changes: Vec<ChangeWire>,
427 blob_dir: Option<&Path>,
428) -> Result<(i64, Vec<PushAck>)> {
429 if changes.is_empty() {
430 return Ok((0, Vec::new()));
431 }
432 let mut tx = begin_immediate(conn).await?;
433 let existing_change_ids = load_existing_change_ids(&mut tx, &changes).await?;
434 let unassigned_changes = changes
435 .iter()
436 .filter(|change| !existing_change_ids.contains(&change.change_id))
437 .cloned()
438 .collect::<Vec<_>>();
439 if unassigned_changes
440 .iter()
441 .any(|change| change.op_type == op_type::ATTACHMENT_ADD)
442 && blob_dir.is_none()
443 {
444 bail!("error attachment-blob-storage-required");
445 }
446 if let Some(blob_dir) = blob_dir {
447 ensure_attachment_blobs_present(&mut tx, blob_dir, &unassigned_changes).await?;
448 }
449 let mut next_server_seq = next_available_server_seq(&mut tx).await?;
450 let mut accepted_count = 0_i64;
451 let mut push_acks = Vec::with_capacity(changes.len());
452 for change in changes {
453 let existing_server_seq = sqlx::query_scalar::<_, Option<i64>>(
454 "SELECT server_seq FROM changes WHERE change_id = ?",
455 )
456 .bind(&change.change_id)
457 .fetch_optional(&mut *tx)
458 .await?
459 .flatten();
460 let server_seq = if let Some(server_seq) = existing_server_seq {
461 verify_existing_change(&mut tx, &change).await?;
462 server_seq
463 } else {
464 let server_seq = next_server_seq;
465 next_server_seq += 1;
466 let payload = change.payload.to_string();
467 sqlx::query(
468 "INSERT INTO changes(change_id, client_id, local_seq, entity_type, entity_id, field,
469 op_type, payload, base_version, created_at, server_seq)
470 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
471 )
472 .bind(&change.change_id)
473 .bind(&change.client_id)
474 .bind(change.local_seq)
475 .bind(&change.entity_type)
476 .bind(&change.entity_id)
477 .bind(&change.field)
478 .bind(&change.op_type)
479 .bind(payload)
480 .bind(&change.base_version)
481 .bind(&change.created_at)
482 .bind(server_seq)
483 .execute(&mut *tx)
484 .await?;
485 apply_server_blob_reference(&mut tx, &change).await?;
486 accepted_count += 1;
487 server_seq
488 };
489 push_acks.push(PushAck {
490 change_id: change.change_id,
491 server_seq,
492 });
493 }
494 tx.commit().await?;
495 Ok((accepted_count, push_acks))
496}
497
498async fn next_available_server_seq(conn: &mut SqliteConnection) -> Result<i64> {
499 Ok(sqlx::query_scalar!(
500 r#"SELECT COALESCE(MAX(server_seq), 0) + 1 AS "seq!: i64" FROM changes"#
501 )
502 .fetch_one(&mut *conn)
503 .await?)
504}
505
506async fn apply_server_blob_reference(
507 conn: &mut SqliteConnection,
508 change: &ChangeWire,
509) -> Result<()> {
510 match change.op_type.as_str() {
511 op_type::ATTACHMENT_ADD => {
512 let workspace_id = change.payload["workspace_id"]
513 .as_str()
514 .context("payload missing workspace_id")?;
515 let attachment_id = change.payload["attachment_id"]
516 .as_str()
517 .context("payload missing attachment_id")?;
518 let sha256 = change.payload["sha256"]
519 .as_str()
520 .context("payload missing sha256")?;
521 let byte_size = change.payload["byte_size"]
522 .as_i64()
523 .context("payload missing byte_size")?;
524 sqlx::query(
525 "INSERT INTO server_blob_references(
526 workspace_id, attachment_id, task_id, sha256, byte_size, deleted
527 ) VALUES (?, ?, ?, ?, ?, 0)
528 ON CONFLICT(workspace_id, attachment_id) DO UPDATE SET
529 task_id = excluded.task_id, sha256 = excluded.sha256,
530 byte_size = excluded.byte_size, deleted = 0",
531 )
532 .bind(workspace_id)
533 .bind(attachment_id)
534 .bind(&change.entity_id)
535 .bind(sha256)
536 .bind(byte_size)
537 .execute(&mut *conn)
538 .await?;
539 sqlx::query(
540 "DELETE FROM blob_upload_reservations WHERE workspace_id = ? AND sha256 = ?",
541 )
542 .bind(workspace_id)
543 .bind(sha256)
544 .execute(&mut *conn)
545 .await?;
546 }
547 op_type::ATTACHMENT_DELETE => {
548 sqlx::query(
549 "UPDATE server_blob_references SET deleted = 1
550 WHERE workspace_id = ? AND attachment_id = ?",
551 )
552 .bind(
553 change.payload["workspace_id"]
554 .as_str()
555 .context("payload missing workspace_id")?,
556 )
557 .bind(
558 change.payload["attachment_id"]
559 .as_str()
560 .context("payload missing attachment_id")?,
561 )
562 .execute(&mut *conn)
563 .await?;
564 }
565 op_type::SET_FIELD if change.field.as_deref() == Some("deleted") => {
566 let deleted = change.payload["value"]
567 .as_str()
568 .is_some_and(|value| value == "1");
569 sqlx::query(
570 "INSERT INTO server_task_tombstones(workspace_id, task_id, deleted)
571 VALUES (?, ?, ?)
572 ON CONFLICT(workspace_id, task_id) DO UPDATE SET deleted = excluded.deleted",
573 )
574 .bind(
575 change.payload["workspace_id"]
576 .as_str()
577 .context("payload missing workspace_id")?,
578 )
579 .bind(&change.entity_id)
580 .bind(i64::from(deleted))
581 .execute(&mut *conn)
582 .await?;
583 }
584 _ => {}
585 }
586 Ok(())
587}
588
589async fn ensure_attachment_blobs_present(
590 conn: &mut SqliteConnection,
591 blob_dir: &Path,
592 changes: &[ChangeWire],
593) -> Result<()> {
594 for change in changes {
595 if change.op_type != op_type::ATTACHMENT_ADD {
596 continue;
597 }
598 let contract = super::blob::attachment_blob_contract(change)?
599 .context("error attachment-blob-missing")?;
600 let Some(row) =
601 crate::attachments::storage::blob_inventory_row(conn, &contract.sha256).await?
602 else {
603 bail!("error attachment-blob-missing");
604 };
605 if !row.available || !crate::attachments::object_path(blob_dir, &contract.sha256)?.exists()
606 {
607 bail!("error attachment-blob-missing");
608 }
609 if row.byte_size != contract.byte_size || row.media_type != contract.media_type {
610 bail!("error blob-inventory-metadata-mismatch");
611 }
612 let bytes =
613 tokio::fs::read(crate::attachments::object_path(blob_dir, &contract.sha256)?).await?;
614 if i64::try_from(bytes.len()).ok() != Some(contract.byte_size)
615 || crate::attachments::storage::sha256_hex(&bytes) != contract.sha256
616 {
617 bail!("error attachment-blob-content-mismatch");
618 }
619 let validated =
620 crate::attachments::decode::validate_image(bytes, Some(contract.media_type.clone()))
621 .await?;
622 if (validated.facts.width, validated.facts.height) != (contract.width, contract.height) {
623 bail!("error blob-inventory-metadata-mismatch");
624 }
625 let admitted: bool = sqlx::query_scalar(
626 "SELECT EXISTS(
627 SELECT 1 FROM server_blob_references sbr
628 LEFT JOIN server_task_tombstones st
629 ON st.workspace_id = sbr.workspace_id AND st.task_id = sbr.task_id
630 WHERE sbr.workspace_id = ? AND sbr.sha256 = ? AND sbr.deleted = 0
631 AND COALESCE(st.deleted, 0) = 0
632 ) OR EXISTS(
633 SELECT 1 FROM blob_upload_reservations
634 WHERE workspace_id = ? AND sha256 = ? AND byte_size = ? AND expires_at > ?
635 )",
636 )
637 .bind(&contract.workspace_id)
638 .bind(&contract.sha256)
639 .bind(&contract.workspace_id)
640 .bind(&contract.sha256)
641 .bind(contract.byte_size)
642 .bind(crate::ids::now())
643 .fetch_one(&mut *conn)
644 .await?;
645 if !admitted {
646 bail!("error attachment-blob-unreserved");
647 }
648 }
649 Ok(())
650}
651
652async fn load_server_changes_after(
653 conn: &mut SqliteConnection,
654 after: i64,
655 pull_limit: u32,
656) -> Result<(Vec<ChangeWire>, bool)> {
657 let fetch_limit = i64::from(pull_limit) + 1;
658 let rows = sqlx::query_as!(
659 ChangeRow,
660 r#"SELECT change_id AS "change_id!: String", client_id AS "client_id!: String",
661 local_seq AS "local_seq!: i64", entity_type AS "entity_type!: String",
662 entity_id AS "entity_id!: String", field, op_type AS "op_type!: String",
663 payload AS "payload!: String", base_version, created_at AS "created_at!: String",
664 server_seq
665 FROM changes WHERE server_seq > ? ORDER BY server_seq LIMIT ?"#,
666 after,
667 fetch_limit,
668 )
669 .fetch_all(&mut *conn)
670 .await?;
671 let has_more = rows.len() > pull_limit as usize;
672 let changes = rows
673 .into_iter()
674 .take(pull_limit as usize)
675 .map(ChangeRow::into_wire)
676 .collect();
677 Ok((changes, has_more))
678}