diff --git a/src/storage/sqlite.rs b/src/storage/sqlite.rs index 17ab1f9..1f9fb59 100644 --- a/src/storage/sqlite.rs +++ b/src/storage/sqlite.rs @@ -9,11 +9,13 @@ use async_trait::async_trait; use bytes::Bytes; use http::HeaderMap; +use regex::Regex; use rusqlite::{params, Connection, OptionalExtension}; +use scraper::Html; use sha2::{Digest, Sha256}; use std::collections::BTreeMap; use std::path::{Path, PathBuf}; -use std::sync::mpsc as std_mpsc; +use std::sync::{mpsc as std_mpsc, LazyLock}; use std::thread; use std::time::Duration; use tokio::sync::mpsc; @@ -2074,6 +2076,52 @@ fn headers_to_json(h: &HeaderMap) -> String { serde_json::to_string(&pairs).unwrap_or_else(|_| "[]".into()) } +fn content_identity_from_html(html: &str, url: &Url) -> String { + const EMPTY_EXCLUDE_TAGS: &[&str] = &[]; + let cleaned = crate::extract::html_clean::clean_html( + html, + &crate::extract::html_clean::CleanOptions { + url: url.as_str(), + exclude_tags: EMPTY_EXCLUDE_TAGS, + only_main_content: true, + }, + ) + .unwrap_or_else(|_| html.to_string()); + let document = Html::parse_document(&cleaned); + let text = document.root_element().text().collect::>().join(" "); + content_identity_from_text(&text) +} + +fn content_identity_from_text(input: &str) -> String { + static TOKEN_KV: LazyLock = LazyLock::new(|| { + Regex::new( + r"\b(?:csrf|xsrf|nonce|session(?:[_ -]?id)?|sid|token|timestamp|ts|request[_ -]?id)\s*[:=]\s*[a-z0-9._:/+-]{6,}\b", + ) + .expect("static regex") + }); + static ISO_TIMESTAMP: LazyLock = LazyLock::new(|| { + Regex::new(r"\b\d{4}-\d{2}-\d{2}[t ][0-9:.+-]+z?\b").expect("static regex") + }); + static UUID: LazyLock = LazyLock::new(|| { + Regex::new(r"\b[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\b") + .expect("static regex") + }); + static LONG_HEX: LazyLock = + LazyLock::new(|| Regex::new(r"\b[0-9a-f]{16,}\b").expect("static regex")); + static LONG_OPAQUE: LazyLock = + LazyLock::new(|| Regex::new(r"\b[a-z0-9_-]{24,}\b").expect("static regex")); + static WS: LazyLock = LazyLock::new(|| Regex::new(r"\s+").expect("static regex")); + + let lower = input.to_ascii_lowercase(); + let without_pairs = TOKEN_KV.replace_all(&lower, " volatile "); + let without_timestamps = ISO_TIMESTAMP.replace_all(&without_pairs, " timestamp "); + let without_uuids = UUID.replace_all(&without_timestamps, " opaque "); + let without_hex = LONG_HEX.replace_all(&without_uuids, " opaque "); + let without_opaque = LONG_OPAQUE.replace_all(&without_hex, " opaque "); + let compact = WS.replace_all(without_opaque.trim(), " "); + hex::encode(Sha256::digest(compact.as_bytes())) +} + async fn write_blob( root: PathBuf, kind: &'static str, @@ -2131,8 +2179,14 @@ impl ArtifactStorage for SqliteStorage { body: &Bytes, truncated: bool, ) -> Result<()> { - let hash = hex::encode(Sha256::digest(body)); let body_vec = body.to_vec(); + let hash = if crate::cache_validator::looks_like_html(headers, &body_vec) { + let (html, _charset) = + crate::impersonate::decode::decode_html_to_string(headers, &body_vec); + content_identity_from_html(&html, final_url) + } else { + hex::encode(Sha256::digest(&body_vec)) + }; let body_size = body_vec.len() as i64; let head_fingerprint = crate::cache_validator::looks_like_html(headers, &body_vec) .then(|| String::from_utf8_lossy(&body_vec).to_string()) @@ -2189,7 +2243,7 @@ impl ArtifactStorage for SqliteStorage { html_post_js: &str, meta: &PageMetadata, ) -> Result<()> { - let hash = hex::encode(Sha256::digest(html_post_js.as_bytes())); + let hash = content_identity_from_html(html_post_js, &meta.final_url); let html_bytes = html_post_js.as_bytes().to_vec(); let blob_size = html_bytes.len() as i64; let blob_path = if self.content_store_enabled { diff --git a/tests/artifact_storage.rs b/tests/artifact_storage.rs index a1a95f0..abb8295 100644 --- a/tests/artifact_storage.rs +++ b/tests/artifact_storage.rs @@ -225,6 +225,143 @@ async fn sqlite_content_store_writes_blobs_without_legacy_inline_columns_by_defa } } +#[tokio::test] +async fn sqlite_content_identity_dedupes_cross_url_without_dropping_edges() { + let tmp = tempfile::tempdir().unwrap(); + let db_path = tmp.path().join("content-id.db"); + let sq = crawlex::storage::sqlite::SqliteStorage::open(&db_path).unwrap(); + + let from = Url::parse("https://sqlite.example/source").unwrap(); + let first = Url::parse("https://sqlite.example/a").unwrap(); + let second = Url::parse("https://sqlite.example/b").unwrap(); + let mut headers = HeaderMap::new(); + headers.insert("content-type", "text/html; charset=utf-8".parse().unwrap()); + let first_body = Bytes::from_static( + br#"Noise A
+

Stable article

The real content is the same.

+

csrf=abcdef1234567890 timestamp=2026-07-11T21:00:00Z

+
"#, + ); + let second_body = Bytes::from_static( + br#"Noise B
+

Stable article

The real content is the same.

+

csrf=fedcba0987654321 timestamp=2026-07-11T21:01:09Z

+
"#, + ); + + sq.save_raw_response(&first, &first, 200, &headers, &first_body, false) + .await + .unwrap(); + sq.save_raw_response(&second, &second, 200, &headers, &second_body, false) + .await + .unwrap(); + sq.save_edge(&from, &second).await.unwrap(); + + let mut rows = Vec::new(); + for _ in 0..20 { + let conn = rusqlite::Connection::open(&db_path).unwrap(); + rows = conn + .prepare( + "SELECT url, body_sha256, body_blob_path FROM pages + WHERE url IN (?1, ?2) ORDER BY url", + ) + .unwrap() + .query_map([first.as_str(), second.as_str()], |r| { + Ok(( + r.get::<_, String>(0)?, + r.get::<_, String>(1)?, + r.get::<_, String>(2)?, + )) + }) + .unwrap() + .filter_map(|r| r.ok()) + .collect::>(); + if rows.len() == 2 { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + + assert_eq!(rows.len(), 2, "both URL rows must be retained"); + assert_eq!(rows[0].1, rows[1].1, "normalised content identity"); + assert_eq!(rows[0].2, rows[1].2, "duplicate content reuses one blob"); + let conn = rusqlite::Connection::open(&db_path).unwrap(); + let blob_count: i64 = conn + .query_row( + "SELECT COUNT(*) FROM content_blobs WHERE kind='raw'", + [], + |r| r.get(0), + ) + .unwrap(); + assert_eq!(blob_count, 1, "body bytes should be stored once"); + let edge_count: i64 = conn + .query_row( + "SELECT COUNT(*) FROM edges WHERE src=?1 AND dst=?2", + [from.as_str(), second.as_str()], + |r| r.get(0), + ) + .unwrap(); + assert_eq!(edge_count, 1, "dedup must not suppress graph edges"); +} + +#[tokio::test] +async fn sqlite_same_url_unchanged_content_identity_short_circuits_blob_store() { + let tmp = tempfile::tempdir().unwrap(); + let db_path = tmp.path().join("same-url.db"); + let sq = crawlex::storage::sqlite::SqliteStorage::open(&db_path).unwrap(); + + let url = Url::parse("https://sqlite.example/page").unwrap(); + let mut headers = HeaderMap::new(); + headers.insert("content-type", "text/html; charset=utf-8".parse().unwrap()); + let first_body = Bytes::from_static( + br#"

Stable page

Same copy.

+

session_id=aaaaaaaaaaaaaaaa timestamp=2026-07-11T21:00:00Z

"#, + ); + let second_body = Bytes::from_static( + br#"

Stable page

Same copy.

+

session_id=bbbbbbbbbbbbbbbb timestamp=2026-07-11T22:30:00Z

"#, + ); + + sq.save_raw_response(&url, &url, 200, &headers, &first_body, false) + .await + .unwrap(); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + let conn = rusqlite::Connection::open(&db_path).unwrap(); + let first_row: (String, String) = conn + .query_row( + "SELECT body_sha256, body_blob_path FROM pages WHERE url=?1", + [url.as_str()], + |r| Ok((r.get(0)?, r.get(1)?)), + ) + .unwrap(); + + sq.save_raw_response(&url, &url, 200, &headers, &second_body, false) + .await + .unwrap(); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + let conn = rusqlite::Connection::open(&db_path).unwrap(); + let second_row: (String, String) = conn + .query_row( + "SELECT body_sha256, body_blob_path FROM pages WHERE url=?1", + [url.as_str()], + |r| Ok((r.get(0)?, r.get(1)?)), + ) + .unwrap(); + let blob_count: i64 = conn + .query_row( + "SELECT COUNT(*) FROM content_blobs WHERE kind='raw'", + [], + |r| r.get(0), + ) + .unwrap(); + + assert_eq!(first_row, second_row); + assert_eq!( + blob_count, 1, + "unchanged content must not create a new blob" + ); +} + #[tokio::test] async fn sqlite_page_cache_metadata_round_trips_validators_and_head_fingerprint() { let tmp = tempfile::tempdir().unwrap();