1use gwk_domain::blob::BlobAddress;
32use gwk_domain::checkpoint::{CHECKPOINT_SCHEMA_VERSION, Checkpoint};
33use gwk_domain::envelope::PayloadRef;
34use gwk_domain::ids::{ByteCount, Seq, Timestamp};
35use gwk_domain::port::BlobStore;
36use gwk_domain::protocol::{ProjectionKind, ProjectionRecord};
37use sha2::{Digest, Sha256};
38use sqlx::{PgConnection, Row};
39
40use crate::blob::container;
41use crate::blob::store::PgBlobStore;
42use crate::numeric::{from_numeric_text, to_numeric_text};
43use crate::project::Refusal;
44
45pub const RECORDS_MEDIA_TYPE: &str = "application/x-ndjson";
51
52struct Projection {
89 tag: &'static str,
90 key: &'static str,
96 query: &'static str,
97 read: &'static str,
107 derived: bool,
109}
110
111const fn derived(
112 tag: &'static str,
113 key: &'static str,
114 query: &'static str,
115 read: &'static str,
116) -> Projection {
117 Projection {
118 tag,
119 key,
120 query,
121 read,
122 derived: true,
123 }
124}
125
126const fn written_beside_the_log(
127 tag: &'static str,
128 key: &'static str,
129 query: &'static str,
130 read: &'static str,
131) -> Projection {
132 Projection {
133 tag,
134 key,
135 query,
136 read,
137 derived: false,
138 }
139}
140
141pub fn read_query(kind: ProjectionKind) -> Option<(&'static str, &'static str)> {
150 PROJECTIONS
151 .iter()
152 .find(|p| p.tag == kind.as_str())
153 .map(|p| (p.read, p.key))
154}
155
156const PROJECTIONS: &[Projection] = &[
157 derived(
158 "attempt",
159 "id",
160 "SELECT jsonb_build_object('projection_type', 'attempt', 'attempt', to_jsonb(t))::text \
161 FROM gwk.attempt t ORDER BY t.id",
162 "SELECT jsonb_build_object('projection_type', 'attempt', 'attempt', to_jsonb(t))::text \
163 FROM gwk.attempt t \
164 WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
165 AND ($2::text IS NULL OR t.id = $2) \
166 ORDER BY t.id COLLATE \"C\" LIMIT $3",
167 ),
168 written_beside_the_log(
169 "attention_item",
170 "id",
171 "SELECT jsonb_build_object('projection_type', 'attention_item', 'attention_item', \
172 to_jsonb(t))::text FROM gwk.attention_item t ORDER BY t.id",
173 "SELECT jsonb_build_object('projection_type', 'attention_item', 'attention_item', \
174 to_jsonb(t))::text FROM gwk.attention_item t \
175 WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
176 AND ($2::text IS NULL OR t.id = $2) \
177 ORDER BY t.id COLLATE \"C\" LIMIT $3",
178 ),
179 derived(
180 "authority_grant",
181 "id",
182 "SELECT jsonb_build_object('projection_type', 'authority_grant', 'authority_grant', \
183 to_jsonb(t))::text FROM gwk.authority_grant t ORDER BY t.id",
184 "SELECT jsonb_build_object('projection_type', 'authority_grant', 'authority_grant', \
185 to_jsonb(t))::text FROM gwk.authority_grant t \
186 WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
187 AND ($2::text IS NULL OR t.id = $2) \
188 ORDER BY t.id COLLATE \"C\" LIMIT $3",
189 ),
190 derived(
191 "command",
192 "id",
193 "SELECT jsonb_build_object('projection_type', 'command', 'command', to_jsonb(t))::text \
194 FROM gwk.command t ORDER BY t.id",
195 "SELECT jsonb_build_object('projection_type', 'command', 'command', to_jsonb(t))::text \
196 FROM gwk.command t \
197 WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
198 AND ($2::text IS NULL OR t.id = $2) \
199 ORDER BY t.id COLLATE \"C\" LIMIT $3",
200 ),
201 derived(
202 "dispatch_node",
203 "id",
204 "SELECT jsonb_build_object('projection_type', 'dispatch_node', 'dispatch_node', \
205 to_jsonb(t))::text FROM gwk.dispatch_node t ORDER BY t.id",
206 "SELECT jsonb_build_object('projection_type', 'dispatch_node', 'dispatch_node', \
207 to_jsonb(t))::text FROM gwk.dispatch_node t \
208 WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
209 AND ($2::text IS NULL OR t.id = $2) \
210 ORDER BY t.id COLLATE \"C\" LIMIT $3",
211 ),
212 derived(
213 "engine_session",
214 "id",
215 "SELECT jsonb_build_object('projection_type', 'engine_session', 'engine_session', \
216 to_jsonb(t))::text FROM gwk.engine_session t ORDER BY t.id",
217 "SELECT jsonb_build_object('projection_type', 'engine_session', 'engine_session', \
218 to_jsonb(t))::text FROM gwk.engine_session t \
219 WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
220 AND ($2::text IS NULL OR t.id = $2) \
221 ORDER BY t.id COLLATE \"C\" LIMIT $3",
222 ),
223 derived(
224 "evidence",
225 "id",
226 "SELECT jsonb_build_object('projection_type', 'evidence', 'evidence', \
227 to_jsonb(t) || jsonb_build_object('byte_size', t.byte_size::text))::text \
228 FROM gwk.evidence t ORDER BY t.id",
229 "SELECT jsonb_build_object('projection_type', 'evidence', 'evidence', \
230 to_jsonb(t) || jsonb_build_object('byte_size', t.byte_size::text))::text \
231 FROM gwk.evidence t \
232 WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
233 AND ($2::text IS NULL OR t.id = $2) \
234 ORDER BY t.id COLLATE \"C\" LIMIT $3",
235 ),
236 derived(
237 "gate",
238 "id",
239 "SELECT jsonb_build_object('projection_type', 'gate', 'gate', to_jsonb(t))::text \
240 FROM gwk.gate t ORDER BY t.id",
241 "SELECT jsonb_build_object('projection_type', 'gate', 'gate', to_jsonb(t))::text \
242 FROM gwk.gate t \
243 WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
244 AND ($2::text IS NULL OR t.id = $2) \
245 ORDER BY t.id COLLATE \"C\" LIMIT $3",
246 ),
247 derived(
248 "ingested_record",
249 "id",
250 "SELECT jsonb_build_object('projection_type', 'ingested_record', 'ingested_record', \
251 to_jsonb(t) || jsonb_build_object('event_seq', t.event_seq::text))::text \
252 FROM gwk.ingested_record t ORDER BY t.id",
253 "SELECT jsonb_build_object('projection_type', 'ingested_record', 'ingested_record', \
254 to_jsonb(t) || jsonb_build_object('event_seq', t.event_seq::text))::text \
255 FROM gwk.ingested_record t \
256 WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
257 AND ($2::text IS NULL OR t.id = $2) \
258 ORDER BY t.id COLLATE \"C\" LIMIT $3",
259 ),
260 derived(
261 "lease",
262 "id",
263 "SELECT jsonb_build_object('projection_type', 'lease', 'lease', \
264 to_jsonb(t) || jsonb_build_object('fence_token', t.fence_token::text))::text \
265 FROM gwk.lease t ORDER BY t.id",
266 "SELECT jsonb_build_object('projection_type', 'lease', 'lease', \
267 to_jsonb(t) || jsonb_build_object('fence_token', t.fence_token::text))::text \
268 FROM gwk.lease t \
269 WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
270 AND ($2::text IS NULL OR t.id = $2) \
271 ORDER BY t.id COLLATE \"C\" LIMIT $3",
272 ),
273 derived(
274 "message",
275 "id",
276 "SELECT jsonb_build_object('projection_type', 'message', 'message', to_jsonb(t))::text \
277 FROM gwk.message t ORDER BY t.id",
278 "SELECT jsonb_build_object('projection_type', 'message', 'message', to_jsonb(t))::text \
279 FROM gwk.message t \
280 WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
281 AND ($2::text IS NULL OR t.id = $2) \
282 ORDER BY t.id COLLATE \"C\" LIMIT $3",
283 ),
284 derived(
285 "orchestrator_checkpoint",
286 "orchestrator_id",
287 "SELECT jsonb_build_object('projection_type', 'orchestrator_checkpoint', \
288 'orchestrator_checkpoint', \
289 (to_jsonb(t) - 'updated_at') || jsonb_build_object('seq', t.seq::text))::text \
290 FROM gwk.orchestrator_checkpoint t ORDER BY t.orchestrator_id",
291 "SELECT jsonb_build_object('projection_type', 'orchestrator_checkpoint', \
295 'orchestrator_checkpoint', \
296 (to_jsonb(t) - 'updated_at') || jsonb_build_object('seq', t.seq::text))::text \
297 FROM gwk.orchestrator_checkpoint t \
298 WHERE ($1::text IS NULL OR t.orchestrator_id COLLATE \"C\" > $1) \
299 AND ($2::text IS NULL OR t.orchestrator_id = $2) \
300 ORDER BY t.orchestrator_id COLLATE \"C\" LIMIT $3",
301 ),
302 written_beside_the_log(
303 "receipt",
304 "id",
305 "SELECT jsonb_build_object('projection_type', 'receipt', 'receipt', \
306 (to_jsonb(t) - 'from_state' - 'to_state') \
307 || jsonb_strip_nulls(jsonb_build_object('from', t.from_state, 'to', t.to_state)))::text \
308 FROM gwk.receipt t ORDER BY t.id",
309 "SELECT jsonb_build_object('projection_type', 'receipt', 'receipt', \
310 (to_jsonb(t) - 'from_state' - 'to_state') \
311 || jsonb_strip_nulls(jsonb_build_object('from', t.from_state, 'to', t.to_state)))::text \
312 FROM gwk.receipt t \
313 WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
314 AND ($2::text IS NULL OR t.id = $2) \
315 ORDER BY t.id COLLATE \"C\" LIMIT $3",
316 ),
317 derived(
318 "task",
319 "id",
320 "SELECT jsonb_build_object('projection_type', 'task', 'task', to_jsonb(t))::text \
321 FROM gwk.task t ORDER BY t.id",
322 "SELECT jsonb_build_object('projection_type', 'task', 'task', to_jsonb(t))::text \
323 FROM gwk.task t \
324 WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
325 AND ($2::text IS NULL OR t.id = $2) \
326 ORDER BY t.id COLLATE \"C\" LIMIT $3",
327 ),
328 derived(
329 "worktree",
330 "id",
331 "SELECT jsonb_build_object('projection_type', 'worktree', 'worktree', to_jsonb(t))::text \
332 FROM gwk.worktree t ORDER BY t.id",
333 "SELECT jsonb_build_object('projection_type', 'worktree', 'worktree', to_jsonb(t))::text \
334 FROM gwk.worktree t \
335 WHERE ($1::text IS NULL OR t.id COLLATE \"C\" > $1) \
336 AND ($2::text IS NULL OR t.id = $2) \
337 ORDER BY t.id COLLATE \"C\" LIMIT $3",
338 ),
339];
340
341pub async fn canonical_records(conn: &mut PgConnection) -> Result<Vec<u8>, Refusal> {
348 records(conn, false).await
349}
350
351pub async fn derived_records(conn: &mut PgConnection) -> Result<Vec<u8>, Refusal> {
359 records(conn, true).await
360}
361
362async fn records(conn: &mut PgConnection, derived_only: bool) -> Result<Vec<u8>, Refusal> {
363 let mut out = Vec::new();
364 for projection in PROJECTIONS {
365 if derived_only && !projection.derived {
366 continue;
367 }
368 let rows = sqlx::query(projection.query)
371 .fetch_all(&mut *conn)
372 .await
373 .map_err(|e| Refusal::storage(format!("read {} projections: {e}", projection.tag)))?;
374 for row in &rows {
375 let raw: String = row
376 .try_get(0)
377 .map_err(|e| Refusal::storage(format!("projection row: {e}")))?;
378 let record: ProjectionRecord = serde_json::from_str(&raw).map_err(|e| {
382 Refusal::storage(format!(
383 "projection row does not match the contract type: {e}"
384 ))
385 })?;
386 serde_json::to_writer(&mut out, &record)
387 .map_err(|e| Refusal::storage(format!("serialize projection record: {e}")))?;
388 out.push(b'\n');
389 }
390 }
391 Ok(out)
392}
393
394pub fn projection_hash(records: &[u8]) -> String {
396 let digest: [u8; 32] = Sha256::digest(records).into();
397 container::hex_lower(&digest)
398}
399
400pub async fn snapshot(
412 conn: &mut PgConnection,
413 blobs: &PgBlobStore,
414 through: Seq,
415 created_at: &Timestamp,
416) -> Result<Checkpoint, Refusal> {
417 let records = derived_records(conn).await?;
418 let hash = projection_hash(&records);
419 let address =
420 BlobAddress::from_digest(&hash).map_err(|e| Refusal::storage(format!("hash: {e}")))?;
421
422 let byte_size = ByteCount::new(records.len() as u64);
423 let store_blob = async {
424 let upload = blobs
425 .begin(RECORDS_MEDIA_TYPE.to_owned(), byte_size)
426 .await?;
427 for (sequence, chunk) in records
428 .chunks(gwk_domain::blob::BLOB_CHUNK_BYTES)
429 .enumerate()
430 {
431 let sequence = u32::try_from(sequence).map_err(|_| {
432 gwk_domain::port::BlobError::Storage("snapshot has too many chunks".to_owned())
433 })?;
434 blobs.write_chunk(&upload, sequence, chunk).await?;
435 }
436 if records.is_empty() {
437 blobs.write_chunk(&upload, 0, &[]).await?;
440 }
441 blobs.commit(upload, address.clone()).await
442 };
443 let (descriptor, _deduped) = store_blob
444 .await
445 .map_err(|e| Refusal::storage(format!("store checkpoint records: {e}")))?;
446 debug_assert_eq!(descriptor.address, address);
451
452 let checkpoint = Checkpoint {
453 schema_version: CHECKPOINT_SCHEMA_VERSION,
454 through_sequence: through,
455 projection_hash: hash,
456 records_ref: PayloadRef {
457 digest: address.as_str().to_owned(),
458 media_type: RECORDS_MEDIA_TYPE.to_owned(),
459 byte_size,
460 retention_class: None,
461 evidence_pin: None,
462 },
463 created_at: created_at.clone(),
464 };
465
466 sqlx::query(
470 "INSERT INTO gwk_internal.checkpoint \
471 (through_seq, schema_version, projection_hash, records_ref, created_at) \
472 VALUES ($1::numeric, $2, $3, $4, $5::timestamptz) \
473 ON CONFLICT (through_seq) DO NOTHING",
474 )
475 .bind(to_numeric_text(through.value()))
476 .bind(i64::from(checkpoint.schema_version))
477 .bind(&checkpoint.projection_hash)
478 .bind(
479 serde_json::to_value(&checkpoint.records_ref)
480 .map_err(|e| Refusal::storage(format!("serialize records_ref: {e}")))?,
481 )
482 .bind(created_at.as_str())
483 .execute(&mut *conn)
484 .await
485 .map_err(|e| Refusal::storage(format!("record checkpoint: {e}")))?;
486
487 sqlx::query(
491 "UPDATE gwk_internal.writer SET checkpoint_seq = $1::numeric, checkpoint_at = $2::timestamptz \
492 WHERE id = 1",
493 )
494 .bind(to_numeric_text(through.value()))
495 .bind(created_at.as_str())
496 .execute(&mut *conn)
497 .await
498 .map_err(|e| Refusal::storage(format!("advance the checkpoint barrier: {e}")))?;
499
500 Ok(checkpoint)
501}
502
503pub async fn checkpoints(conn: &mut PgConnection) -> Result<Vec<Checkpoint>, Refusal> {
505 let rows = sqlx::query(
506 "SELECT through_seq::text AS through_text, schema_version, projection_hash, records_ref, \
507 to_json(created_at) #>> '{}' AS created_at \
508 FROM gwk_internal.checkpoint ORDER BY through_seq DESC",
509 )
510 .fetch_all(conn)
511 .await
512 .map_err(|e| Refusal::storage(format!("read checkpoints: {e}")))?;
513
514 rows.iter()
515 .map(|row| {
516 let get = |name: &str| -> Result<String, Refusal> {
517 row.try_get(name)
518 .map_err(|e| Refusal::storage(format!("column {name}: {e}")))
519 };
520 let schema_version: i64 = row
521 .try_get("schema_version")
522 .map_err(|e| Refusal::storage(format!("column schema_version: {e}")))?;
523 let records_ref: serde_json::Value = row
524 .try_get("records_ref")
525 .map_err(|e| Refusal::storage(format!("column records_ref: {e}")))?;
526 Ok(Checkpoint {
527 schema_version: u32::try_from(schema_version)
528 .map_err(|e| Refusal::storage(format!("schema_version: {e}")))?,
529 through_sequence: Seq::new(
530 from_numeric_text(&get("through_text")?)
531 .map_err(|e| Refusal::storage(format!("column through_seq: {e}")))?,
532 ),
533 projection_hash: get("projection_hash")?,
534 records_ref: serde_json::from_value(records_ref)
535 .map_err(|e| Refusal::storage(format!("column records_ref: {e}")))?,
536 created_at: Timestamp::new(get("created_at")?),
537 })
538 })
539 .collect()
540}
541
542#[cfg(test)]
543mod tests {
544 use super::*;
545
546 #[test]
547 fn every_projection_is_visited_exactly_once_in_a_written_down_order() {
548 let ordered: Vec<&str> = PROJECTIONS.iter().map(|p| p.tag).collect();
553 let mut tags = ordered.clone();
554 tags.sort_unstable();
555 tags.dedup();
556 assert_eq!(tags.len(), PROJECTIONS.len(), "a tag appears twice");
557 assert_eq!(ordered, tags, "the visit order must be alphabetical");
558
559 for projection in PROJECTIONS {
560 let tag = projection.tag;
561 assert!(
567 projection
568 .query
569 .contains(&format!("'projection_type', '{tag}', '{tag}',")),
570 "{tag}: the record's one field must be named for its tag"
571 );
572 assert!(
573 projection.query.contains(&format!(" FROM gwk.{tag} t ")),
574 "{tag}: the query must read the table it is tagged for"
575 );
576 assert!(
579 projection.query.contains(" ORDER BY "),
580 "{tag}: rows must be ordered"
581 );
582 }
583 }
584
585 #[test]
586 fn only_the_tables_a_replay_can_rebuild_reach_the_hash() {
587 let excluded: Vec<&str> = PROJECTIONS
592 .iter()
593 .filter(|p| !p.derived)
594 .map(|p| p.tag)
595 .collect();
596 assert_eq!(excluded, ["attention_item", "receipt"]);
597 assert_eq!(
598 PROJECTIONS.iter().filter(|p| p.derived).count(),
599 PROJECTIONS.len() - 2,
600 "everything else must be rebuildable from the log"
601 );
602 }
603
604 #[test]
605 fn the_hash_is_over_the_stored_bytes_and_nothing_else() {
606 let records = b"{\"projection_type\":\"task\"}\n".to_vec();
609 let hash = projection_hash(&records);
610 let address = BlobAddress::from_digest(&hash).expect("a legal address");
611 assert_eq!(address.digest_hex(), hash);
612 assert_eq!(
613 hash,
614 {
615 let digest: [u8; 32] = Sha256::digest(&records).into();
616 container::hex_lower(&digest)
617 },
618 "the hash must be a plain SHA-256 over the bytes"
619 );
620 assert_eq!(
622 projection_hash(&[]),
623 "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
624 );
625 }
626
627 #[test]
628 fn a_served_row_is_the_same_row_the_hash_canonicalizes() {
629 for projection in PROJECTIONS {
635 let head = |q: &'static str| {
636 q.split_once(" FROM ")
637 .expect("every projection query selects FROM a table")
638 .0
639 .to_owned()
640 };
641 assert_eq!(
642 head(projection.query),
643 head(projection.read),
644 "{} builds a different record for the hash than for a read",
645 projection.tag
646 );
647 }
648 }
649
650 #[test]
651 fn the_cursor_key_is_the_column_the_page_was_ordered_by() {
652 for projection in PROJECTIONS {
657 let key = projection.key;
658 assert!(
659 projection
660 .read
661 .contains(&format!("ORDER BY t.{key} COLLATE")),
662 "{} pages by {key} but does not order by it",
663 projection.tag
664 );
665 assert!(
666 projection
667 .read
668 .contains(&format!("t.{key} COLLATE \"C\" > $1")),
669 "{} orders by {key} but compares the cursor against another column",
670 projection.tag
671 );
672 }
673 }
674
675 #[test]
676 fn every_projection_a_client_can_name_has_a_read_behind_it() {
677 for kind in ProjectionKind::ALL {
681 assert!(
682 read_query(*kind).is_some(),
683 "{} has no read query",
684 kind.as_str()
685 );
686 }
687 }
688}