chore: #64 Content identity + non-destructive content dedup (ADR-0006)

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::<Vec<_>>().join(" ");
+    content_identity_from_text(&text)
+}
+
+fn content_identity_from_text(input: &str) -> String {
+    static TOKEN_KV: LazyLock<Regex> = 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<Regex> = LazyLock::new(|| {
+        Regex::new(r"\b\d{4}-\d{2}-\d{2}[t ][0-9:.+-]+z?\b").expect("static regex")
+    });
+    static UUID: LazyLock<Regex> = 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<Regex> =
+        LazyLock::new(|| Regex::new(r"\b[0-9a-f]{16,}\b").expect("static regex"));
+    static LONG_OPAQUE: LazyLock<Regex> =
+        LazyLock::new(|| Regex::new(r"\b[a-z0-9_-]{24,}\b").expect("static regex"));
+    static WS: LazyLock<Regex> = 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 {

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#"<html><head><title>Noise A</title></head><body><main>
+            <h1>Stable article</h1><p>The real content is the same.</p>
+            <p>csrf=abcdef1234567890 timestamp=2026-07-11T21:00:00Z</p>
+        </main></body></html>"#,
+    );
+    let second_body = Bytes::from_static(
+        br#"<html><head><title>Noise B</title></head><body><main>
+            <h1>Stable article</h1><p>The real content is the same.</p>
+            <p>csrf=fedcba0987654321 timestamp=2026-07-11T21:01:09Z</p>
+        </main></body></html>"#,
+    );
+
+    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::<Vec<_>>();
+        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#"<html><body><main><h1>Stable page</h1><p>Same copy.</p>
+        <p>session_id=aaaaaaaaaaaaaaaa timestamp=2026-07-11T21:00:00Z</p></main></body></html>"#,
+    );
+    let second_body = Bytes::from_static(
+        br#"<html><body><main><h1>Stable page</h1><p>Same copy.</p>
+        <p>session_id=bbbbbbbbbbbbbbbb timestamp=2026-07-11T22:30:00Z</p></main></body></html>"#,
+    );
+
+    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();