1use crate::auth::{AnonymousAuth, DynAuthProvider, StaticTokenAuth};
45use crate::embed::{embed_one, Embedder};
46use crate::error::{Error, Result};
47use crate::extract::{AutoExtractor, Descriptor, Extractor};
48use crate::fetch::{FetchOptions, Fetcher, HttpFetcher};
49use crate::index::{Index, IndexBuilder, UpsertOutcome};
50use crate::query::{PreparedQuery, RankParams};
51use crate::resource::{Resource, ResourceKind, SourceRef};
52use crate::source::{CrawlScope, CrawlSource, Source};
53use crate::url_key::UrlKeyExt as _;
54use crate::{text, Filter};
55use bytes::Bytes;
56use compact_str::CompactString;
57use futures::stream::{FuturesUnordered, StreamExt};
58use graph_r::hash::xxh3_64;
59use std::path::{Path, PathBuf};
60use std::sync::OnceLock;
61use std::time::Duration;
62
63#[cfg(feature = "onnx")]
65pub type DefaultEmbedder = crate::embed::OnnxEmbedder;
66#[cfg(not(feature = "onnx"))]
68pub type DefaultEmbedder = crate::embed::HashEmbedder;
69
70#[cfg(not(feature = "onnx"))]
72const HASH_DIM: usize = 256;
73
74const DEFAULT_EMBED_BATCH: usize = 64;
76const REFRESH_MAX_BYTES: u64 = 2 * 1024 * 1024;
78
79#[cfg(feature = "onnx")]
85pub fn default_embedder() -> Result<DefaultEmbedder> {
86 crate::embed::OnnxEmbedder::new()
87}
88#[cfg(not(feature = "onnx"))]
95#[allow(clippy::unnecessary_wraps)] pub fn default_embedder() -> Result<DefaultEmbedder> {
97 Ok(crate::embed::HashEmbedder::new(HASH_DIM))
98}
99
100use graph_r::fsio::now_ms;
101
102#[derive(Clone, Copy, Debug)]
104pub struct ExportDoc<'a> {
105 pub meta: &'a crate::index::DocMeta,
107 pub edges: &'a [crate::url_key::UrlKey],
109}
110
111#[derive(Clone, Debug)]
113pub struct SearchResult {
114 pub url: String,
116 pub score: f32,
118 pub title: Option<String>,
120 pub snippet: String,
122 pub kind: ResourceKind,
124 pub tags: Vec<String>,
126}
127
128#[derive(Clone, Debug, Default, PartialEq, Eq)]
130pub struct UpdateReport {
131 pub added: usize,
133 pub updated: usize,
135 pub unchanged: usize,
137 pub skipped: usize,
139 pub failed: usize,
141 pub pages: Vec<PageOutcome>,
145}
146
147#[derive(Clone, Copy, Debug, PartialEq, Eq)]
149pub enum PageChange {
150 Added,
152 Updated,
154 Unchanged,
156 Skipped,
158 Removed,
160}
161
162#[derive(Clone, Debug, PartialEq, Eq)]
164pub struct PageOutcome {
165 pub url: String,
167 pub url_key: crate::url_key::UrlKey,
169 pub change: PageChange,
171 pub title: Option<CompactString>,
173 pub headings: Vec<(u8, CompactString)>,
175 pub keywords: Vec<CompactString>,
177}
178
179impl UpdateReport {
180 #[must_use]
182 pub fn pages_seen(&self) -> usize {
183 self.added + self.updated + self.unchanged + self.skipped
184 }
185}
186
187#[derive(Clone, Debug, Default, PartialEq, Eq)]
189pub struct RefreshReport {
190 pub refreshed: usize,
192 pub unchanged: usize,
194 pub removed: usize,
196 pub failed: usize,
198 pub pages: Vec<PageOutcome>,
200}
201
202impl RefreshReport {
203 #[must_use]
205 pub fn total(&self) -> usize {
206 self.refreshed + self.unchanged + self.removed + self.failed
207 }
208}
209
210#[derive(Debug)]
221pub struct LinkIndex<E = DefaultEmbedder> {
222 path: Option<PathBuf>,
223 embedder: E,
224 opened: Option<Index>,
226 builder: Option<IndexBuilder>,
228 cached: OnceLock<Index>,
230 http: OnceLock<reqwest::Client>,
232}
233
234impl LinkIndex {
235 pub fn open_or_create(path: impl Into<PathBuf>) -> Result<Self> {
241 Self::open_or_create_with(path, default_embedder()?)
242 }
243
244 pub fn open(path: impl Into<PathBuf>) -> Result<Self> {
246 Self::open_with(path, default_embedder()?)
247 }
248
249 pub fn in_memory() -> Result<Self> {
251 Ok(Self::in_memory_with(default_embedder()?))
252 }
253
254 pub fn from_bytes(bytes: Bytes) -> Result<Self> {
262 Self::from_bytes_with(bytes, default_embedder()?)
263 }
264}
265
266impl<E: Embedder> LinkIndex<E> {
267 fn from_builder(path: Option<PathBuf>, embedder: E, builder: IndexBuilder) -> Self {
268 Self {
269 path,
270 embedder,
271 opened: None,
272 builder: Some(builder),
273 cached: OnceLock::new(),
274 http: OnceLock::new(),
275 }
276 }
277
278 fn from_opened(path: Option<PathBuf>, embedder: E, opened: Index) -> Self {
279 Self {
280 path,
281 embedder,
282 opened: Some(opened),
283 builder: None,
284 cached: OnceLock::new(),
285 http: OnceLock::new(),
286 }
287 }
288
289 pub fn open_or_create_with(path: impl Into<PathBuf>, embedder: E) -> Result<Self> {
294 let path = path.into();
295 if path.exists() {
296 let index =
297 check_compatible(Index::open(&path)?, &embedder, &path.display().to_string())?;
298 Ok(Self::from_opened(Some(path), embedder, index))
299 } else {
300 let builder =
301 IndexBuilder::new(embedder.dim(), embedder.metric(), embedder.identity().raw());
302 Ok(Self::from_builder(Some(path), embedder, builder))
303 }
304 }
305
306 pub fn open_with(path: impl Into<PathBuf>, embedder: E) -> Result<Self> {
311 let path = path.into();
312 if !path.exists() {
313 return Err(Error::not_found(path.display().to_string()));
314 }
315 let index = check_compatible(Index::open(&path)?, &embedder, &path.display().to_string())?;
316 Ok(Self::from_opened(Some(path), embedder, index))
317 }
318
319 #[must_use]
321 pub fn in_memory_with(embedder: E) -> Self {
322 let builder =
323 IndexBuilder::new(embedder.dim(), embedder.metric(), embedder.identity().raw());
324 Self::from_builder(None, embedder, builder)
325 }
326
327 pub fn from_bytes_with(bytes: Bytes, embedder: E) -> Result<Self> {
332 let index = check_compatible(Index::from_bytes(bytes)?, &embedder, "in-memory image")?;
333 Ok(Self::from_opened(None, embedder, index))
334 }
335
336 pub fn reload_from_bytes(&mut self, bytes: Bytes) -> Result<()> {
345 let index = check_compatible(Index::from_bytes(bytes)?, &self.embedder, "in-memory image")?;
346 self.opened = Some(index);
347 self.builder = None;
348 self.cached = OnceLock::new();
349 Ok(())
350 }
351
352 #[must_use]
354 pub fn embedder(&self) -> &E {
355 &self.embedder
356 }
357
358 pub fn to_bytes(&self) -> Result<Vec<u8>> {
364 match (&self.opened, &self.builder) {
365 (Some(idx), _) => idx.serialize(),
366 (None, Some(b)) => Ok(b.serialize()),
367 (None, None) => Err(Error::format("index has no content to serialize")),
368 }
369 }
370
371 pub fn freeze(&mut self) -> Result<()> {
379 let Some(builder) = self.builder.take() else {
380 return Ok(());
381 };
382 let image = Bytes::from(builder.serialize());
383 match Index::from_bytes(image) {
384 Ok(index) => {
385 self.opened = Some(index);
386 self.cached = OnceLock::new();
387 Ok(())
388 }
389 Err(e) => {
390 self.builder = Some(builder);
391 Err(e)
392 }
393 }
394 }
395
396 fn http_client(&self) -> Result<reqwest::Client> {
398 if let Some(c) = self.http.get() {
399 return Ok(c.clone());
400 }
401 let c = crate::fetch::http::default_client()?;
402 Ok(self.http.get_or_init(|| c).clone())
403 }
404
405 #[must_use]
407 pub fn len(&self) -> usize {
408 match (&self.opened, &self.builder) {
409 (Some(idx), _) => idx.len(),
410 (None, Some(b)) => b.len(),
411 (None, None) => 0,
412 }
413 }
414
415 #[must_use]
417 pub fn is_empty(&self) -> bool {
418 self.len() == 0
419 }
420
421 pub fn update(&mut self, url: impl Into<String>) -> UpdateBuilder<'_, E> {
424 UpdateBuilder::<'_, E>::new(self, url.into())
425 }
426
427 pub fn refresh(&mut self) -> RefreshBuilder<'_, E> {
430 RefreshBuilder::<'_, E>::new(self)
431 }
432
433 #[cfg(feature = "github")]
439 pub fn update_github(&mut self, spec: impl AsRef<str>) -> Result<GithubUpdateBuilder<'_, E>> {
440 let spec = crate::source::github::GithubSpec::parse(spec.as_ref())?;
441 Ok(GithubUpdateBuilder::<'_, E>::new(self, spec))
442 }
443
444 pub fn pin(&mut self, url_prefix: &str) -> Result<usize> {
447 Ok(self.materialize_builder()?.set_pinned(url_prefix, true))
448 }
449
450 pub fn unpin(&mut self, url_prefix: &str) -> Result<usize> {
452 Ok(self.materialize_builder()?.set_pinned(url_prefix, false))
453 }
454
455 pub async fn search(&self, query: &str, limit: usize) -> Result<Vec<SearchResult>> {
460 self.search_filtered(query, limit, &Filter::All).await
461 }
462
463 pub async fn search_filtered(
469 &self,
470 query: &str,
471 limit: usize,
472 filter: &Filter,
473 ) -> Result<Vec<SearchResult>> {
474 self.search_ranked(query, limit, filter, 0.0).await
475 }
476
477 pub async fn search_ranked(
485 &self,
486 query: &str,
487 limit: usize,
488 filter: &Filter,
489 graph_boost: f32,
490 ) -> Result<Vec<SearchResult>> {
491 let vector = embed_one(&self.embedder, query).await?;
492 let terms: Vec<CompactString> = text::normalized_tokens(query).collect();
493 let index = self.read_index();
494 let pq = PreparedQuery {
495 vector: &vector,
496 terms: &terms,
497 filter,
498 limit,
499 rank: RankParams {
500 graph_boost,
501 ..RankParams::default()
502 },
503 };
504 let hits = index.search_prepared(&pq)?;
505 Ok(hits.iter().map(SearchResult::from_hit).collect())
506 }
507
508 pub fn related(&self, url: &str, k: usize) -> Result<Vec<SearchResult>> {
514 let key = crate::url_key::UrlKey::parse(url)?;
515 let index = self.read_index();
516 Ok(index
517 .related(key, k)
518 .iter()
519 .map(SearchResult::from_hit)
520 .collect())
521 }
522
523 #[allow(clippy::missing_errors_doc)]
527 pub fn export(&self) -> Result<impl Iterator<Item = ExportDoc<'_>>> {
528 let idx = self.read_index();
529 Ok(idx
530 .documents()
531 .iter()
532 .zip(idx.edge_lists())
533 .map(|(meta, edges)| ExportDoc { meta, edges }))
534 }
535
536 pub fn save(&self) -> Result<()> {
542 let path = self
543 .path
544 .as_ref()
545 .ok_or_else(|| Error::invalid_url("no path bound; use save_as"))?;
546 self.save_to(path)
547 }
548
549 pub fn save_as(&mut self, path: impl Into<PathBuf>) -> Result<()> {
551 let path = path.into();
552 self.save_to(&path)?;
553 self.path = Some(path);
554 Ok(())
555 }
556
557 fn save_to(&self, path: &Path) -> Result<()> {
558 match (&self.opened, &self.builder) {
559 (Some(idx), _) => idx.save(path),
560 (None, Some(b)) => b.save(path),
561 (None, None) => Err(Error::format("index has no content to save")),
562 }
563 }
564
565 fn read_index(&self) -> &Index {
568 if let Some(idx) = &self.opened {
569 return idx;
570 }
571 self.cached.get_or_init(|| {
572 self.builder
573 .as_ref()
574 .expect("opened or builder is always present")
575 .to_index()
576 })
577 }
578
579 fn materialize_builder(&mut self) -> Result<&mut IndexBuilder> {
582 if self.builder.is_none() {
583 let idx = self
584 .opened
585 .take()
586 .ok_or_else(|| Error::format("index has no builder or opened state"))?;
587 self.builder = Some(idx.into_builder()?);
588 }
589 self.cached = OnceLock::new(); Ok(self.builder.as_mut().expect("just materialized"))
591 }
592
593 pub async fn ingest_from<S: Source>(
601 &mut self,
602 source: &S,
603 root: &SourceRef,
604 embed_batch: usize,
605 pin: bool,
606 ) -> Result<UpdateReport> {
607 self.materialize_builder()?;
608 let dim = self.embedder.dim();
609 let embed_batch = embed_batch.max(1);
610 let extractor = AutoExtractor;
611 let LinkIndex {
612 builder,
613 embedder,
614 cached,
615 ..
616 } = self;
617 let embedder: &E = embedder;
618 let builder = builder.as_mut().expect("materialized");
619
620 let mut report = UpdateReport::default();
621 let mut buf = vec![0.0f32; embed_batch * dim];
622 let mut pending: Vec<(Resource, Descriptor, u64)> = Vec::with_capacity(embed_batch);
623
624 let stream = source.discover(root);
625 futures::pin_mut!(stream);
626 while let Some(item) = stream.next().await {
627 let page = item?;
628 let resource = page.resource;
629 let bytes = page.payload.into_bytes().await?;
630 let content_hash = xxh3_64(&bytes);
631 if builder.is_unchanged(&resource.url, content_hash) {
632 report.unchanged += 1;
633 report
634 .pages
635 .push(page_outcome(&resource.url, PageChange::Unchanged, None));
636 continue;
637 }
638 let descriptor = extractor.extract(&resource, &bytes)?;
639 if descriptor.is_empty() {
640 report.skipped += 1;
641 report
642 .pages
643 .push(page_outcome(&resource.url, PageChange::Skipped, None));
644 continue;
645 }
646 pending.push((resource, descriptor, content_hash));
647 if pending.len() == embed_batch {
648 flush_batch(
649 builder,
650 embedder,
651 &mut pending,
652 &mut buf,
653 dim,
654 pin,
655 &mut report,
656 )
657 .await?;
658 }
659 }
660 flush_batch(
661 builder,
662 embedder,
663 &mut pending,
664 &mut buf,
665 dim,
666 pin,
667 &mut report,
668 )
669 .await?;
670
671 if report.added + report.updated > 0 {
672 *cached = OnceLock::new();
673 }
674 Ok(report)
675 }
676
677 #[allow(clippy::too_many_lines)] pub async fn refresh_with<F: Fetcher>(
687 &mut self,
688 fetcher: &F,
689 opts: RefreshOptions,
690 ) -> Result<RefreshReport> {
691 self.materialize_builder()?;
692 let now = now_ms();
693 let dim = self.embedder.dim();
694 let extractor = AutoExtractor;
695
696 let mut evict: Vec<url::Url> = Vec::new();
698 let mut stale: Vec<(url::Url, Option<CompactString>)> = Vec::new();
699 {
700 let builder = self.builder.as_ref().expect("materialized");
701 for d in builder.documents() {
702 if let Some(filter) = &opts.urls {
703 if !filter.contains(&d.url_key) {
704 continue;
705 }
706 }
707 let age = now.saturating_sub(d.fetched_at_ms);
708 if !d.pinned {
709 if let Some(max_age) = opts.max_age {
710 if u128::from(age) > max_age.as_millis() {
711 if let Ok(url) = url::Url::parse(&d.url) {
712 evict.push(url);
713 }
714 continue; }
716 }
717 }
718 if u128::from(age) >= opts.ttl.as_millis() {
719 if let Ok(url) = url::Url::parse(&d.url) {
720 stale.push((url, d.etag.clone()));
721 }
722 }
723 }
724 }
725
726 let mut report = RefreshReport::default();
727 let builder = self.builder.as_mut().expect("materialized");
728 let pinned_keys: std::collections::HashSet<crate::url_key::UrlKey> = builder
730 .documents()
731 .iter()
732 .filter(|d| d.pinned)
733 .map(|d| d.url_key)
734 .collect();
735
736 for url in &evict {
738 if builder.remove(url) {
739 report.removed += 1;
740 }
741 }
742
743 let mut pending_changed: Vec<(Resource, Descriptor, u64)> = Vec::new();
745 let concurrency = opts.concurrency.max(1);
746 let mut idx = 0;
747 let mut in_flight = FuturesUnordered::new();
748 loop {
749 while in_flight.len() < concurrency && idx < stale.len() {
750 let (url, etag) = stale[idx].clone();
751 idx += 1;
752 in_flight.push(conditional_fetch(fetcher, url, etag));
753 }
754 let Some((url, outcome)) = in_flight.next().await else {
755 break;
756 };
757 match outcome {
758 Ok(RefreshFetch::NotModified) => {
759 builder.touch(&url, now);
760 report.unchanged += 1;
761 report
762 .pages
763 .push(page_outcome(&url, PageChange::Unchanged, None));
764 }
765 Ok(RefreshFetch::Changed { kind, etag, bytes }) => {
766 let content_hash = xxh3_64(&bytes);
767 let mut resource = Resource::new(url.clone()).with_kind(kind);
768 if let Some(tag) = etag {
769 resource = resource.with_etag(tag);
770 }
771 let descriptor = extractor.extract(&resource, &bytes)?;
772 if descriptor.is_empty() {
773 builder.touch(&url, now);
774 report.unchanged += 1;
775 report
776 .pages
777 .push(page_outcome(&url, PageChange::Unchanged, None));
778 continue;
779 }
780 pending_changed.push((resource, descriptor, content_hash));
784 }
785 Err(RefreshError::Gone) => {
786 let pinned = pinned_keys.contains(&crate::url_key::UrlKey::from_url(&url));
788 if opts.evict_unreachable && !pinned && builder.remove(&url) {
789 report.removed += 1;
790 report
791 .pages
792 .push(page_outcome(&url, PageChange::Removed, None));
793 } else {
794 report.failed += 1;
795 }
796 }
797 Err(RefreshError::Transient) => report.failed += 1,
798 }
799 }
800
801 let mut buf = vec![0.0f32; DEFAULT_EMBED_BATCH * dim];
804 while !pending_changed.is_empty() {
805 let take = pending_changed.len().min(DEFAULT_EMBED_BATCH);
806 let batch: Vec<(Resource, Descriptor, u64)> = pending_changed.drain(..take).collect();
807 let texts: Vec<&str> = batch
808 .iter()
809 .map(|(_, d, _)| d.embed_text.as_str())
810 .collect();
811 let out = &mut buf[..texts.len() * dim];
812 self.embedder.embed_batch(&texts, out).await?;
813 drop(texts);
814 for (i, (resource, descriptor, content_hash)) in batch.into_iter().enumerate() {
815 let vector = out[i * dim..(i + 1) * dim].to_vec();
816 let outcome_meta = descriptor_outcome_meta(&descriptor);
817 let mut doc =
818 descriptor.into_document(resource.url, resource.kind, content_hash, vector);
819 doc.fetched_at_ms = now;
820 doc.etag = resource.etag;
821 let url = doc.url.clone();
822 if builder.upsert(doc)? == UpsertOutcome::Unchanged {
827 builder.touch(&url, now);
828 report.unchanged += 1;
829 report
830 .pages
831 .push(page_outcome(&url, PageChange::Unchanged, None));
832 } else {
833 report.refreshed += 1;
834 report
835 .pages
836 .push(page_outcome(&url, PageChange::Updated, Some(outcome_meta)));
837 }
838 }
839 }
840
841 if report.refreshed + report.removed > 0 {
842 self.cached = OnceLock::new();
843 }
844 Ok(report)
845 }
846}
847
848impl SearchResult {
849 fn from_hit(h: crate::query::Hit<'_>) -> Self {
850 Self {
851 url: h.url.to_owned(),
852 score: h.score,
853 title: h.title.map(str::to_owned),
854 snippet: h.snippet.to_owned(),
855 kind: h.kind,
856 tags: h.tags().map(str::to_owned).collect(),
857 }
858 }
859}
860
861async fn flush_batch<E: Embedder>(
863 builder: &mut IndexBuilder,
864 embedder: &E,
865 pending: &mut Vec<(Resource, Descriptor, u64)>,
866 buf: &mut [f32],
867 dim: usize,
868 pin: bool,
869 report: &mut UpdateReport,
870) -> Result<()> {
871 if pending.is_empty() {
872 return Ok(());
873 }
874 let now = now_ms();
875 let texts: Vec<&str> = pending
876 .iter()
877 .map(|(_, d, _)| d.embed_text.as_str())
878 .collect();
879 let out = &mut buf[..texts.len() * dim];
880 embedder.embed_batch(&texts, out).await?;
881 drop(texts);
882
883 for (i, (resource, descriptor, content_hash)) in pending.drain(..).enumerate() {
884 let vector = out[i * dim..(i + 1) * dim].to_vec();
885 let outcome_meta = descriptor_outcome_meta(&descriptor);
886 let mut doc = descriptor.into_document(resource.url, resource.kind, content_hash, vector);
887 doc.fetched_at_ms = now;
888 doc.etag = resource.etag;
889 doc.pinned = pin;
890 let url = doc.url.clone();
891 let change = match builder.upsert(doc)? {
892 crate::index::UpsertOutcome::Added => {
893 report.added += 1;
894 PageChange::Added
895 }
896 crate::index::UpsertOutcome::Updated => {
897 report.updated += 1;
898 PageChange::Updated
899 }
900 crate::index::UpsertOutcome::Unchanged => {
901 report.unchanged += 1;
902 PageChange::Unchanged
903 }
904 };
905 report
906 .pages
907 .push(page_outcome(&url, change, Some(outcome_meta)));
908 }
909 Ok(())
910}
911
912type OutcomeMeta = (
915 Option<CompactString>,
916 Vec<(u8, CompactString)>,
917 Vec<CompactString>,
918);
919
920fn descriptor_outcome_meta(d: &Descriptor) -> OutcomeMeta {
921 let headings = d
922 .headings
923 .iter()
924 .enumerate()
925 .map(|(i, h)| {
926 let level = d
927 .heading_levels
928 .get(i)
929 .copied()
930 .unwrap_or((i.min(2) + 1) as u8);
931 (level, h.clone())
932 })
933 .collect();
934 (
935 d.title.clone(),
936 headings,
937 d.keywords.iter().cloned().collect(),
938 )
939}
940
941fn page_outcome(url: &url::Url, change: PageChange, meta: Option<OutcomeMeta>) -> PageOutcome {
942 let (title, headings, keywords) = meta.unwrap_or_default();
943 PageOutcome {
944 url: crate::url_key::canonicalize(url),
945 url_key: crate::url_key::UrlKey::from_url(url),
946 change,
947 title,
948 headings,
949 keywords,
950 }
951}
952
953enum RefreshFetch {
955 NotModified,
956 Changed {
957 kind: ResourceKind,
958 etag: Option<CompactString>,
959 bytes: Bytes,
960 },
961}
962
963enum RefreshError {
965 Gone,
967 Transient,
969}
970
971async fn conditional_fetch<F: Fetcher>(
974 fetcher: &F,
975 url: url::Url,
976 etag: Option<CompactString>,
977) -> (url::Url, std::result::Result<RefreshFetch, RefreshError>) {
978 let resource = Resource::new(url);
979 let opts = FetchOptions {
980 if_none_match: etag.as_deref(),
981 max_bytes: Some(REFRESH_MAX_BYTES),
982 user_agent: None,
983 };
984 let outcome = match fetcher.fetch(&resource, opts).await {
985 Ok(got) => {
986 let kind = got.meta.kind;
987 let etag = got.meta.etag;
988 match got.payload.into_bytes().await {
989 Ok(bytes) => Ok(RefreshFetch::Changed { kind, etag, bytes }),
990 Err(_) => Err(RefreshError::Transient),
991 }
992 }
993 Err(Error::NotModified { .. }) => Ok(RefreshFetch::NotModified),
994 Err(
995 Error::NotFound { .. } | Error::PermissionDenied { .. } | Error::Unauthenticated { .. },
996 ) => Err(RefreshError::Gone),
997 Err(_) => Err(RefreshError::Transient),
998 };
999 (resource.url, outcome)
1000}
1001
1002fn check_compatible<E: Embedder>(index: Index, embedder: &E, what: &str) -> Result<Index> {
1003 if index.dim() != embedder.dim() || index.embedder_id() != embedder.identity().raw() {
1004 return Err(Error::embed(format!(
1005 "index at {what} is incompatible with the current embedder (index dim {}, embedder dim {})",
1006 index.dim(),
1007 embedder.dim()
1008 )));
1009 }
1010 Ok(index)
1011}
1012
1013#[derive(Clone, Debug)]
1015pub struct RefreshOptions {
1016 pub urls: Option<std::collections::HashSet<crate::url_key::UrlKey>>,
1019 pub ttl: Duration,
1021 pub max_age: Option<Duration>,
1023 pub evict_unreachable: bool,
1025 pub concurrency: usize,
1027}
1028
1029impl Default for RefreshOptions {
1030 fn default() -> Self {
1031 Self {
1032 urls: None,
1033 ttl: Duration::from_secs(24 * 60 * 60),
1034 max_age: None,
1035 evict_unreachable: true,
1036 concurrency: 8,
1037 }
1038 }
1039}
1040
1041pub struct UpdateBuilder<'a, E = DefaultEmbedder> {
1043 index: &'a mut LinkIndex<E>,
1044 root: String,
1045 depth: u16,
1046 scope: CrawlScope,
1047 max_pages: usize,
1048 concurrency: usize,
1049 max_retries: u32,
1050 min_delay: Duration,
1051 embed_batch: usize,
1052 pin: bool,
1053 require_path: Vec<CompactString>,
1054 accept_extensions: Vec<CompactString>,
1055 index_path: Vec<CompactString>,
1056 #[cfg(feature = "robots")]
1057 respect_robots: bool,
1058 auth: Box<dyn DynAuthProvider>,
1059 extra_validators: Vec<(crate::url_key::UrlKey, CompactString)>,
1065 extra_known_edges: Vec<(crate::url_key::UrlKey, Vec<url::Url>)>,
1066}
1067
1068impl<E> std::fmt::Debug for UpdateBuilder<'_, E> {
1069 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1070 f.debug_struct("UpdateBuilder")
1071 .field("root", &self.root)
1072 .field("depth", &self.depth)
1073 .field("max_pages", &self.max_pages)
1074 .field("concurrency", &self.concurrency)
1075 .finish_non_exhaustive()
1076 }
1077}
1078
1079impl<'a, E: Embedder> UpdateBuilder<'a, E> {
1080 fn new(index: &'a mut LinkIndex<E>, root: String) -> Self {
1081 Self {
1082 index,
1083 root,
1084 depth: 2,
1085 scope: CrawlScope::PathPrefix,
1086 max_pages: 1000,
1087 concurrency: 8,
1088 max_retries: 2,
1089 min_delay: Duration::ZERO,
1090 embed_batch: DEFAULT_EMBED_BATCH,
1091 pin: false,
1092 require_path: Vec::new(),
1093 accept_extensions: Vec::new(),
1094 index_path: Vec::new(),
1095 #[cfg(feature = "robots")]
1096 respect_robots: false,
1097 auth: Box::new(AnonymousAuth),
1098 extra_validators: Vec::new(),
1099 extra_known_edges: Vec::new(),
1100 }
1101 }
1102
1103 #[must_use]
1105 pub fn depth(mut self, depth: u16) -> Self {
1106 self.depth = depth;
1107 self
1108 }
1109
1110 #[must_use]
1112 pub fn scope(mut self, scope: CrawlScope) -> Self {
1113 self.scope = scope;
1114 self
1115 }
1116
1117 #[must_use]
1119 pub fn max_pages(mut self, max_pages: usize) -> Self {
1120 self.max_pages = max_pages;
1121 self
1122 }
1123
1124 #[must_use]
1126 pub fn concurrency(mut self, concurrency: usize) -> Self {
1127 self.concurrency = concurrency;
1128 self
1129 }
1130
1131 #[must_use]
1133 pub fn max_retries(mut self, max_retries: u32) -> Self {
1134 self.max_retries = max_retries;
1135 self
1136 }
1137
1138 #[must_use]
1140 pub fn min_delay(mut self, delay: Duration) -> Self {
1141 self.min_delay = delay;
1142 self
1143 }
1144
1145 #[must_use]
1147 pub fn embed_batch(mut self, n: usize) -> Self {
1148 self.embed_batch = n.max(1);
1149 self
1150 }
1151
1152 #[must_use]
1154 pub fn pin(mut self) -> Self {
1155 self.pin = true;
1156 self
1157 }
1158
1159 #[cfg(feature = "robots")]
1161 #[must_use]
1162 pub fn respect_robots(mut self, respect: bool) -> Self {
1163 self.respect_robots = respect;
1164 self
1165 }
1166
1167 #[must_use]
1170 pub fn require_path(mut self, substring: impl Into<CompactString>) -> Self {
1171 self.require_path.push(substring.into());
1172 self
1173 }
1174
1175 #[must_use]
1178 pub fn accept_extension(mut self, extension: impl Into<CompactString>) -> Self {
1179 self.accept_extensions.push(extension.into());
1180 self
1181 }
1182
1183 #[must_use]
1186 pub fn index_path(mut self, substring: impl Into<CompactString>) -> Self {
1187 self.index_path.push(substring.into());
1188 self
1189 }
1190
1191 #[must_use]
1202 pub fn validators(
1203 mut self,
1204 entries: impl IntoIterator<Item = (crate::url_key::UrlKey, CompactString)>,
1205 ) -> Self {
1206 self.extra_validators.extend(entries);
1207 self
1208 }
1209
1210 #[must_use]
1216 pub fn known_edges(
1217 mut self,
1218 entries: impl IntoIterator<Item = (crate::url_key::UrlKey, Vec<url::Url>)>,
1219 ) -> Self {
1220 self.extra_known_edges.extend(entries);
1221 self
1222 }
1223
1224 #[must_use]
1228 pub fn token(mut self, token: impl Into<String>) -> Self {
1229 if let Ok(SourceRef::Http { root }) = SourceRef::http(&self.root) {
1230 if let Some(host) = root.host_str() {
1231 self.auth = Box::new(StaticTokenAuth::bearer_scoped(token, host));
1232 return self;
1233 }
1234 }
1235 self.auth = Box::new(StaticTokenAuth::bearer(token));
1236 self
1237 }
1238
1239 #[must_use]
1241 pub fn auth(mut self, auth: Box<dyn DynAuthProvider>) -> Self {
1242 self.auth = auth;
1243 self
1244 }
1245
1246 pub async fn run(self) -> Result<UpdateReport> {
1252 let root = SourceRef::http(&self.root)?;
1253 self.index.materialize_builder()?;
1257 let (validators, known_edges) = {
1258 let builder = self.index.builder.as_ref().expect("materialized");
1259 let docs = builder.documents();
1260 let by_key: std::collections::HashMap<crate::url_key::UrlKey, &str> =
1261 docs.iter().map(|d| (d.url_key, d.url.as_str())).collect();
1262 let mut validators = self.extra_validators;
1266 let mut known_edges = self.extra_known_edges;
1267 for (d, edges) in docs.iter().zip(builder.edge_lists()) {
1268 if let Some(etag) = &d.etag {
1269 validators.push((d.url_key, etag.clone()));
1270 }
1271 let children: Vec<url::Url> = edges
1272 .iter()
1273 .filter_map(|k| by_key.get(k))
1274 .filter_map(|u| url::Url::parse(u).ok())
1275 .collect();
1276 if !children.is_empty() {
1277 known_edges.push((d.url_key, children));
1278 }
1279 }
1280 (validators, known_edges)
1281 };
1282 let fetcher = HttpFetcher::with_client(self.index.http_client()?, self.auth);
1283 #[allow(unused_mut)]
1284 let mut source = CrawlSource::new(fetcher)
1285 .validators(validators)
1286 .known_edges(known_edges)
1287 .depth(self.depth)
1288 .scope(self.scope)
1289 .max_pages(self.max_pages)
1290 .concurrency(self.concurrency)
1291 .max_retries(self.max_retries)
1292 .min_delay(self.min_delay);
1293 for substring in self.require_path {
1294 source = source.require_path(substring);
1295 }
1296 for ext in self.accept_extensions {
1297 source = source.accept_extension(ext);
1298 }
1299 for substring in self.index_path {
1300 source = source.index_path(substring);
1301 }
1302 #[cfg(feature = "robots")]
1303 {
1304 source = source.respect_robots(self.respect_robots);
1305 }
1306 let mut report = self
1307 .index
1308 .ingest_from(&source, &root, self.embed_batch, self.pin)
1309 .await?;
1310 report.failed = source.failed_count() as usize;
1311 for url in source.take_revalidated() {
1316 report.unchanged += 1;
1317 report
1318 .pages
1319 .push(page_outcome(&url, PageChange::Unchanged, None));
1320 }
1321 Ok(report)
1322 }
1323}
1324
1325#[cfg(feature = "github")]
1328pub struct GithubUpdateBuilder<'a, E = DefaultEmbedder> {
1329 index: &'a mut LinkIndex<E>,
1330 spec: crate::source::github::GithubSpec,
1331 token: Option<String>,
1332 api_base: Option<CompactString>,
1333 raw_base: Option<CompactString>,
1334 depth: Option<u16>,
1335 max_bytes: u64,
1336 concurrency: usize,
1337 embed_batch: usize,
1338 pin: bool,
1339 accept_extensions: Vec<CompactString>,
1340 extra_validators: Vec<(crate::url_key::UrlKey, CompactString)>,
1341}
1342
1343#[cfg(feature = "github")]
1344impl<E> std::fmt::Debug for GithubUpdateBuilder<'_, E> {
1345 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1346 f.debug_struct("GithubUpdateBuilder")
1347 .field("spec", &self.spec)
1348 .field("depth", &self.depth)
1349 .field("has_token", &self.token.is_some())
1350 .finish_non_exhaustive()
1351 }
1352}
1353
1354#[cfg(feature = "github")]
1355impl<'a, E: Embedder> GithubUpdateBuilder<'a, E> {
1356 fn new(index: &'a mut LinkIndex<E>, spec: crate::source::github::GithubSpec) -> Self {
1357 Self {
1358 index,
1359 spec,
1360 token: None,
1361 api_base: None,
1362 raw_base: None,
1363 depth: None,
1364 max_bytes: REFRESH_MAX_BYTES,
1365 concurrency: 8,
1366 embed_batch: DEFAULT_EMBED_BATCH,
1367 pin: false,
1368 accept_extensions: Vec::new(),
1369 extra_validators: Vec::new(),
1370 }
1371 }
1372
1373 #[must_use]
1377 pub fn token(mut self, token: impl Into<String>) -> Self {
1378 self.token = Some(token.into());
1379 self
1380 }
1381
1382 #[must_use]
1384 pub fn bases(
1385 mut self,
1386 api_base: impl Into<CompactString>,
1387 raw_base: impl Into<CompactString>,
1388 ) -> Self {
1389 self.api_base = Some(api_base.into());
1390 self.raw_base = Some(raw_base.into());
1391 self
1392 }
1393
1394 #[must_use]
1396 pub fn depth(mut self, depth: u16) -> Self {
1397 self.depth = Some(depth);
1398 self
1399 }
1400
1401 #[must_use]
1403 pub fn max_bytes(mut self, max_bytes: u64) -> Self {
1404 self.max_bytes = max_bytes;
1405 self
1406 }
1407
1408 #[must_use]
1410 pub fn concurrency(mut self, concurrency: usize) -> Self {
1411 self.concurrency = concurrency.max(1);
1412 self
1413 }
1414
1415 #[must_use]
1417 pub fn embed_batch(mut self, n: usize) -> Self {
1418 self.embed_batch = n.max(1);
1419 self
1420 }
1421
1422 #[must_use]
1424 pub fn pin(mut self) -> Self {
1425 self.pin = true;
1426 self
1427 }
1428
1429 #[must_use]
1431 pub fn accept_extension(mut self, extension: impl AsRef<str>) -> Self {
1432 let e = extension
1433 .as_ref()
1434 .trim_start_matches('.')
1435 .to_ascii_lowercase();
1436 self.accept_extensions.push(CompactString::from(e));
1437 self
1438 }
1439
1440 #[must_use]
1445 pub fn validators(
1446 mut self,
1447 entries: impl IntoIterator<Item = (crate::url_key::UrlKey, CompactString)>,
1448 ) -> Self {
1449 self.extra_validators.extend(entries);
1450 self
1451 }
1452
1453 pub async fn run(self) -> Result<UpdateReport> {
1461 use crate::source::github::{GitHubSource, GithubAuth};
1462
1463 self.index.materialize_builder()?;
1464 let mut validators = self.extra_validators;
1467 {
1468 let builder = self.index.builder.as_ref().expect("materialized");
1469 for d in builder.documents() {
1470 if let Some(etag) = &d.etag {
1471 validators.push((d.url_key, etag.clone()));
1472 }
1473 }
1474 }
1475
1476 let auth = match (&self.api_base, &self.raw_base) {
1477 (Some(api), Some(raw)) => {
1478 let api_host = url::Url::parse(api)
1479 .ok()
1480 .and_then(|u| u.host_str().map(CompactString::from))
1481 .unwrap_or_else(|| CompactString::from("api.github.com"));
1482 let raw_host = url::Url::parse(raw)
1483 .ok()
1484 .and_then(|u| u.host_str().map(CompactString::from))
1485 .unwrap_or_else(|| CompactString::from("raw.githubusercontent.com"));
1486 GithubAuth::new(self.token).with_hosts(api_host, raw_host)
1487 }
1488 _ => GithubAuth::new(self.token),
1489 };
1490 let fetcher = HttpFetcher::with_client(self.index.http_client()?, auth);
1491 let mut source = GitHubSource::new(fetcher, self.spec)
1492 .max_bytes(self.max_bytes)
1493 .concurrency(self.concurrency)
1494 .validators(validators);
1495 if let (Some(api), Some(raw)) = (self.api_base, self.raw_base) {
1496 source = source.bases(api, raw);
1497 }
1498 if let Some(depth) = self.depth {
1499 source = source.depth(depth);
1500 }
1501 for ext in self.accept_extensions {
1502 source = source.accept_extension(ext);
1503 }
1504
1505 let root = SourceRef::http("https://api.github.com/")?;
1508 let mut report = self
1509 .index
1510 .ingest_from(&source, &root, self.embed_batch, self.pin)
1511 .await?;
1512 report.failed = source.failed_count() as usize;
1513 for url in source.take_revalidated() {
1517 report.unchanged += 1;
1518 report
1519 .pages
1520 .push(page_outcome(&url, PageChange::Unchanged, None));
1521 }
1522 Ok(report)
1523 }
1524}
1525
1526pub struct RefreshBuilder<'a, E = DefaultEmbedder> {
1528 index: &'a mut LinkIndex<E>,
1529 opts: RefreshOptions,
1530 auth: Box<dyn DynAuthProvider>,
1531 scoped_host: Option<CompactString>,
1532}
1533
1534impl<E> std::fmt::Debug for RefreshBuilder<'_, E> {
1535 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1536 f.debug_struct("RefreshBuilder")
1537 .field("opts", &self.opts)
1538 .finish_non_exhaustive()
1539 }
1540}
1541
1542impl<'a, E: Embedder> RefreshBuilder<'a, E> {
1543 fn new(index: &'a mut LinkIndex<E>) -> Self {
1544 Self {
1545 index,
1546 opts: RefreshOptions::default(),
1547 auth: Box::new(AnonymousAuth),
1548 scoped_host: None,
1549 }
1550 }
1551
1552 #[must_use]
1554 pub fn ttl(mut self, ttl: Duration) -> Self {
1555 self.opts.ttl = ttl;
1556 self
1557 }
1558
1559 #[must_use]
1562 pub fn max_age(mut self, max_age: Duration) -> Self {
1563 self.opts.max_age = Some(max_age);
1564 self
1565 }
1566
1567 #[must_use]
1569 pub fn evict_unreachable(mut self, evict: bool) -> Self {
1570 self.opts.evict_unreachable = evict;
1571 self
1572 }
1573
1574 #[must_use]
1576 pub fn concurrency(mut self, concurrency: usize) -> Self {
1577 self.opts.concurrency = concurrency;
1578 self
1579 }
1580
1581 #[must_use]
1586 pub fn urls<I>(mut self, urls: I) -> Self
1587 where
1588 I: IntoIterator,
1589 I::Item: AsRef<str>,
1590 {
1591 let set: std::collections::HashSet<crate::url_key::UrlKey> = urls
1592 .into_iter()
1593 .filter_map(|u| url::Url::parse(u.as_ref()).ok())
1594 .map(|u| crate::url_key::UrlKey::from_url(&u))
1595 .collect();
1596 self.opts.urls = Some(set);
1597 self
1598 }
1599
1600 #[must_use]
1603 pub fn token(mut self, token: impl Into<String>, host: impl Into<CompactString>) -> Self {
1604 let host = host.into();
1605 self.auth = Box::new(StaticTokenAuth::bearer_scoped(token, host.as_str()));
1606 self.scoped_host = Some(host);
1607 self
1608 }
1609
1610 #[must_use]
1612 pub fn auth(mut self, auth: Box<dyn DynAuthProvider>) -> Self {
1613 self.auth = auth;
1614 self
1615 }
1616
1617 pub async fn run(self) -> Result<RefreshReport> {
1622 let fetcher = HttpFetcher::with_client(self.index.http_client()?, self.auth);
1623 self.index.refresh_with(&fetcher, self.opts).await
1624 }
1625}
1626
1627#[cfg(test)]
1628mod tests {
1629 use super::*;
1630 use crate::resource::Resource;
1631
1632 #[tokio::test]
1633 async fn in_memory_search_over_manual_upserts() {
1634 let mut idx = LinkIndex::in_memory().unwrap();
1635 index_doc(&mut idx, "https://x.dev/cat", "cats are feline animals").await;
1636 index_doc(&mut idx, "https://x.dev/dog", "dogs are canine animals").await;
1637
1638 let hits = idx.search("canine", 5).await.unwrap();
1639 assert_eq!(hits[0].url, "https://x.dev/dog");
1640 assert_eq!(idx.len(), 2);
1641 }
1642
1643 #[tokio::test]
1644 async fn save_and_reopen_roundtrip() {
1645 let tmp = tempfile::tempdir().unwrap();
1646 let path = tmp.path().join("kb.lnkr");
1647 {
1648 let mut idx = LinkIndex::open_or_create(&path).unwrap();
1649 index_doc(&mut idx, "https://x.dev/a", "alpha content here").await;
1650 idx.save().unwrap();
1651 }
1652 let idx = LinkIndex::open(&path).unwrap();
1653 assert_eq!(idx.len(), 1);
1654 assert!(idx.opened.is_some() && idx.builder.is_none());
1656 let hits = idx.search("alpha", 5).await.unwrap();
1657 assert_eq!(hits[0].url, "https://x.dev/a");
1658 }
1659
1660 #[tokio::test]
1661 async fn search_cache_is_built_and_invalidated() {
1662 let mut idx = LinkIndex::in_memory().unwrap();
1663 index_doc(&mut idx, "https://x.dev/a", "alpha").await;
1664 assert!(idx.cached.get().is_none());
1666 let _ = idx.search("alpha", 5).await.unwrap();
1667 assert!(idx.cached.get().is_some());
1668 index_doc(&mut idx, "https://x.dev/b", "beta").await;
1670 assert!(idx.cached.get().is_none());
1671 let hits = idx.search("beta", 5).await.unwrap();
1672 assert_eq!(hits[0].url, "https://x.dev/b");
1673 }
1674
1675 #[tokio::test]
1676 async fn first_mutation_after_open_materializes_builder() {
1677 let tmp = tempfile::tempdir().unwrap();
1678 let path = tmp.path().join("kb.lnkr");
1679 {
1680 let mut idx = LinkIndex::open_or_create(&path).unwrap();
1681 index_doc(&mut idx, "https://x.dev/a", "alpha").await;
1682 idx.save().unwrap();
1683 }
1684 let mut idx = LinkIndex::open(&path).unwrap();
1685 assert!(idx.opened.is_some());
1686 idx.pin("https://x.dev/").unwrap();
1687 assert!(idx.opened.is_none() && idx.builder.is_some());
1689 assert_eq!(idx.len(), 1);
1690 }
1691
1692 use crate::fetch::{FetchMeta, Fetched};
1693 use crate::payload::DocPayload;
1694 use std::collections::HashMap;
1695 use std::time::Duration;
1696
1697 enum Behavior {
1699 NotModified,
1700 Changed(Vec<u8>),
1701 Gone,
1702 }
1703
1704 struct RefreshMock {
1705 behavior: HashMap<String, Behavior>,
1706 fetched: std::sync::Mutex<Vec<String>>,
1708 }
1709
1710 impl RefreshMock {
1711 fn new(pairs: Vec<(&str, Behavior)>) -> Self {
1712 Self {
1713 behavior: pairs.into_iter().map(|(u, b)| (u.to_string(), b)).collect(),
1714 fetched: std::sync::Mutex::new(Vec::new()),
1715 }
1716 }
1717 }
1718
1719 impl Fetcher for RefreshMock {
1720 type FetchFuture<'a>
1721 = std::future::Ready<Result<Fetched<'a>>>
1722 where
1723 Self: 'a;
1724 fn fetch<'a>(
1725 &'a self,
1726 resource: &'a Resource,
1727 _opts: FetchOptions<'a>,
1728 ) -> Self::FetchFuture<'a> {
1729 let key = crate::url_key::canonicalize(&resource.url);
1730 self.fetched.lock().unwrap().push(key.clone());
1731 let result = match self.behavior.get(&key) {
1732 Some(Behavior::NotModified) => Err(Error::not_modified(resource.url.as_str())),
1733 Some(Behavior::Changed(body)) => Ok(Fetched {
1734 meta: FetchMeta {
1735 kind: ResourceKind::Text,
1736 etag: None,
1737 status: 200,
1738 final_url: None,
1739 },
1740 payload: DocPayload::Owned(Bytes::from(body.clone())),
1741 }),
1742 Some(Behavior::Gone) | None => Err(Error::not_found(resource.url.as_str())),
1743 };
1744 std::future::ready(result)
1745 }
1746 }
1747
1748 #[tokio::test]
1749 async fn refresh_touches_changes_and_evicts() {
1750 let mut idx = LinkIndex::in_memory().unwrap();
1751 index_doc(&mut idx, "https://x.dev/keep", "keep body unchanged").await;
1752 index_doc(&mut idx, "https://x.dev/change", "old body").await;
1753 index_doc(&mut idx, "https://x.dev/dead", "dead body").await;
1754 assert_eq!(idx.len(), 3);
1755
1756 let fetcher = RefreshMock::new(vec![
1757 ("https://x.dev/keep", Behavior::NotModified),
1758 (
1759 "https://x.dev/change",
1760 Behavior::Changed(b"brand new different content".to_vec()),
1761 ),
1762 ("https://x.dev/dead", Behavior::Gone),
1763 ]);
1764 let opts = RefreshOptions {
1765 urls: None,
1766 ttl: Duration::ZERO, max_age: None,
1768 evict_unreachable: true,
1769 concurrency: 4,
1770 };
1771 let report = idx.refresh_with(&fetcher, opts).await.unwrap();
1772 assert_eq!(report.unchanged, 1, "keep was 304");
1773 assert_eq!(report.refreshed, 1, "change re-indexed");
1774 assert_eq!(report.removed, 1, "dead evicted");
1775 assert_eq!(idx.len(), 2);
1776 let hits = idx.search("different", 5).await.unwrap();
1777 assert!(hits.iter().any(|h| h.url == "https://x.dev/change"));
1778 }
1779
1780 #[tokio::test]
1781 async fn refresh_urls_subset_only_touches_listed() {
1782 let mut idx = LinkIndex::in_memory().unwrap();
1783 index_doc(&mut idx, "https://x.dev/one", "first body words").await;
1784 index_doc(&mut idx, "https://x.dev/two", "second body words").await;
1785 index_doc(&mut idx, "https://x.dev/three", "third body words").await;
1786
1787 let fetcher = RefreshMock::new(vec![
1788 ("https://x.dev/one", Behavior::NotModified),
1789 ("https://x.dev/two", Behavior::NotModified),
1790 ("https://x.dev/three", Behavior::NotModified),
1791 ]);
1792 let opts = RefreshOptions {
1793 urls: Some(
1794 [crate::url_key::UrlKey::parse("https://x.dev/two").unwrap()]
1795 .into_iter()
1796 .collect(),
1797 ),
1798 ttl: Duration::ZERO,
1799 max_age: None,
1800 evict_unreachable: true,
1801 concurrency: 4,
1802 };
1803 let report = idx.refresh_with(&fetcher, opts).await.unwrap();
1804 assert_eq!(report.total(), 1, "only the listed URL was acted on");
1805 let requested = fetcher.fetched.lock().unwrap();
1806 assert_eq!(&*requested, &vec!["https://x.dev/two".to_string()]);
1807 }
1808
1809 #[tokio::test]
1810 async fn export_surfaces_documents_and_edges() {
1811 let mut idx = LinkIndex::in_memory().unwrap();
1812 index_doc(
1813 &mut idx,
1814 "https://x.dev/hub",
1815 "<a href=\"/spoke\">s</a> hub body words",
1816 )
1817 .await;
1818 index_doc(&mut idx, "https://x.dev/spoke", "spoke body words").await;
1821
1822 let docs: Vec<_> = idx.export().unwrap().collect();
1823 assert_eq!(docs.len(), 2);
1824 assert!(docs.iter().any(|d| d.meta.url == "https://x.dev/hub"));
1825 assert!(docs
1826 .iter()
1827 .all(|d| d.meta.url_key == crate::url_key::UrlKey::parse(&d.meta.url).unwrap()));
1828 }
1829
1830 #[tokio::test]
1834 async fn refresh_of_identical_body_without_etag_advances_freshness() {
1835 let mut idx = LinkIndex::in_memory().unwrap();
1836 index_doc(&mut idx, "https://x.dev/static", "stable page body words").await;
1837
1838 let fetcher = RefreshMock::new(vec![(
1839 "https://x.dev/static",
1840 Behavior::Changed(b"stable page body words".to_vec()),
1841 )]);
1842
1843 let opts = RefreshOptions {
1845 urls: None,
1846 ttl: Duration::ZERO,
1847 max_age: None,
1848 evict_unreachable: true,
1849 concurrency: 2,
1850 };
1851 let report = idx.refresh_with(&fetcher, opts).await.unwrap();
1852 assert_eq!(report.unchanged, 1, "identical body counts as unchanged");
1853 assert_eq!(report.refreshed, 0);
1854 assert_eq!(fetcher.fetched.lock().unwrap().len(), 1);
1855
1856 let opts = RefreshOptions {
1859 urls: None,
1860 ttl: Duration::from_secs(3600),
1861 max_age: None,
1862 evict_unreachable: true,
1863 concurrency: 2,
1864 };
1865 let report = idx.refresh_with(&fetcher, opts).await.unwrap();
1866 assert_eq!(report.total(), 0, "no work: page is fresh now");
1867 assert_eq!(
1868 fetcher.fetched.lock().unwrap().len(),
1869 1,
1870 "no second network fetch for a fresh page"
1871 );
1872 }
1873
1874 #[tokio::test]
1875 async fn max_age_evicts_unpinned_without_fetching() {
1876 let mut idx = LinkIndex::in_memory().unwrap();
1877 index_doc(&mut idx, "https://x.dev/old", "old").await; let fetcher = RefreshMock::new(vec![]);
1879 let opts = RefreshOptions {
1880 urls: None,
1881 ttl: Duration::from_secs(u64::from(u32::MAX)), max_age: Some(Duration::from_millis(1)), evict_unreachable: true,
1884 concurrency: 4,
1885 };
1886 let report = idx.refresh_with(&fetcher, opts).await.unwrap();
1887 assert_eq!(report.removed, 1);
1888 assert_eq!(idx.len(), 0);
1889 assert!(
1890 fetcher.fetched.lock().unwrap().is_empty(),
1891 "max-age eviction must not fetch"
1892 );
1893 }
1894
1895 #[tokio::test]
1896 async fn graph_related_and_boost_roundtrip() {
1897 use crate::index::{Document, IndexBuilder};
1898 use crate::resource::ResourceKind;
1899 use crate::url_key::UrlKey;
1900
1901 fn node(url: &str, terms: &[&str], targets: &[&str]) -> Document {
1903 let u = url::Url::parse(url).unwrap();
1904 let edges = targets
1905 .iter()
1906 .map(|t| UrlKey::from_url(&url::Url::parse(t).unwrap()))
1907 .collect();
1908 Document {
1909 url: u,
1910 kind: ResourceKind::Html,
1911 content_hash: url.len() as u64,
1912 title: None,
1913 snippet: "s".into(),
1914 lang: None,
1915 tags: smallvec::SmallVec::new(),
1916 terms: terms.iter().map(|t| (*t).into()).collect(),
1917 vector: vec![1.0],
1918 edges,
1919 fetched_at_ms: 0,
1920 etag: None,
1921 pinned: false,
1922 }
1923 }
1924 let mut b = IndexBuilder::new(1, crate::metric::Metric::Cosine, 0);
1925 b.upsert(node(
1926 "https://x.dev/hub",
1927 &["hub"],
1928 &["https://x.dev/a", "https://x.dev/b"],
1929 ))
1930 .unwrap();
1931 b.upsert(node("https://x.dev/a", &["alpha"], &["https://x.dev/b"]))
1932 .unwrap();
1933 b.upsert(node("https://x.dev/b", &["beta"], &[])).unwrap();
1934
1935 let tmp = tempfile::tempdir().unwrap();
1937 let path = tmp.path().join("g.lnkr");
1938 b.save(&path).unwrap();
1939 let index = crate::index::Index::open(&path).unwrap();
1940
1941 let rel = index.related(
1943 UrlKey::from_url(&url::Url::parse("https://x.dev/a").unwrap()),
1944 5,
1945 );
1946 let urls: Vec<&str> = rel.iter().map(|h| h.url).collect();
1947 assert!(urls.contains(&"https://x.dev/b"), "related: {urls:?}");
1948 }
1949
1950 #[tokio::test]
1951 async fn pinned_survives_max_age() {
1952 let mut idx = LinkIndex::in_memory().unwrap();
1953 index_doc(&mut idx, "https://x.dev/pinned", "x").await;
1954 assert_eq!(idx.pin("https://x.dev/pinned").unwrap(), 1);
1955 let fetcher = RefreshMock::new(vec![]);
1956 let opts = RefreshOptions {
1957 urls: None,
1958 ttl: Duration::from_secs(u64::from(u32::MAX)),
1959 max_age: Some(Duration::from_millis(1)),
1960 evict_unreachable: true,
1961 concurrency: 4,
1962 };
1963 let report = idx.refresh_with(&fetcher, opts).await.unwrap();
1964 assert_eq!(report.removed, 0, "pinned exempt from max-age");
1965 assert_eq!(idx.len(), 1);
1966 }
1967
1968 async fn index_doc(idx: &mut LinkIndex, url: &str, body: &str) {
1970 idx.materialize_builder().unwrap(); let resource = Resource::new(url::Url::parse(url).unwrap()).with_kind(ResourceKind::Text);
1972 let descriptor = AutoExtractor.extract(&resource, body.as_bytes()).unwrap();
1973 let vector = embed_one(&idx.embedder, &descriptor.embed_text)
1974 .await
1975 .unwrap();
1976 let doc = descriptor.into_document(
1977 resource.url,
1978 resource.kind,
1979 xxh3_64(body.as_bytes()),
1980 vector,
1981 );
1982 idx.builder.as_mut().unwrap().upsert(doc).unwrap();
1983 }
1984}