Slightly more robust output

Author: GavinMendelGleason

src/indexer.rs

@@ -33,7 +33,7 @@ pub struct IndexPoint {
 }
 
 impl Point {
-    pub fn id(&self) -> &String {
+    pub fn id(&self) -> &str {
         match self {
             Point::Stored { id, vec } => id,
             Point::Mem { vec } => panic!("You can not get the external id of a memory point"),
@@ -100,10 +100,18 @@ pub async fn operations_to_point_operations(
             Operation::Inserted { string, id } => Some((Op::Insert, string.into(), id.into())),
             Operation::Changed { string, id } => Some((Op::Changed, string.into(), id.into())),
             Operation::Deleted { id: _ } => None,
+            Operation::Error { message } => {
+                eprintln!("{}", message);
+                None
+            }
         })
         .collect();
     let strings: Vec<String> = tuples.iter().map(|(_, s, _)| s.to_string()).collect();
-    let vecs: Vec<Embedding> = embeddings_for(key, &strings).await.unwrap();
+    let vecs: Vec<Embedding> = if strings.is_empty() {
+        Vec::new()
+    } else {
+        embeddings_for(key, &strings).await.unwrap()
+    };
     let domain = vector_store.get_domain(domain).unwrap();
     let loaded_vecs = vector_store
         .add_and_load_vecs(&domain, vecs.iter())
@@ -170,12 +178,17 @@ pub enum SearchError {
 
 #[derive(Clone, Debug, PartialEq)]
 pub struct PointQuery {
+    id: usize,
     point: Point,
     distance: u32,
 }
 
 impl PointQuery {
-    pub fn id(&self) -> &String {
+    pub fn internal_id(&self) -> usize {
+        self.id
+    }
+
+    pub fn id(&self) -> &str {
         self.point.id()
     }
 
@@ -197,6 +210,7 @@ pub fn search(p: &Point, num: usize, hnsw: &HnswIndex) -> Result<Vec<PointQuery>
     let mut points = Vec::with_capacity(num);
     for elt in output {
         points.push(PointQuery {
+            id: elt.index,
             point: hnsw.feature(elt.index).clone(),
             distance: elt.distance,
         })

src/server.rs

@@ -48,6 +48,7 @@ pub enum Operation {
     Inserted { string: String, id: String },
     Changed { string: String, id: String },
     Deleted { id: String },
+    Error { message: String },
 }
 
 #[derive(Deserialize, Debug)]
@@ -79,6 +80,11 @@ enum ResourceSpec {
         id: String,
         count: usize,
     },
+    DuplicateCandidates {
+        domain: String,
+        commit: String,
+        threshold: f32,
+    },
 }
 
 #[derive(Debug)]
@@ -104,6 +110,7 @@ fn uri_to_spec(uri: &Uri) -> Result<ResourceSpec, SpecParseError> {
         static ref RE_CHECK: Regex = Regex::new(r"^/check(/?)$").unwrap();
         static ref RE_SEARCH: Regex = Regex::new(r"^/search(/?)$").unwrap();
         static ref RE_SIMILAR: Regex = Regex::new(r"^/similar(/?)$").unwrap();
+        static ref RE_DUPLICATES: Regex = Regex::new(r"^/duplicates(/?)$").unwrap();
     }
     let path = uri.path();
 
@@ -163,6 +170,22 @@ fn uri_to_spec(uri: &Uri) -> Result<ResourceSpec, SpecParseError> {
             }
             _ => Err(SpecParseError::NoCommitIdOrDomain),
         }
+    } else if RE_DUPLICATES.is_match(path) {
+        let query = query_map(uri);
+        let domain = query.get("domain").map(|v| v.to_string());
+        let commit = query.get("commit").map(|v| v.to_string());
+        let threshold = query.get("threshold").map(|v| v.parse::<f32>().unwrap());
+        match (domain, commit) {
+            (Some(domain), Some(commit)) => {
+                let threshold = threshold.unwrap_or(0.0);
+                Ok(ResourceSpec::DuplicateCandidates {
+                    domain,
+                    commit,
+                    threshold,
+                })
+            }
+            _ => Err(SpecParseError::NoCommitIdOrDomain),
+        }
     } else {
         Err(SpecParseError::UnknownPath)
     }
@@ -216,13 +239,20 @@ async fn get_operations_from_terminusdb(
     let lines = StreamReader::new(res).lines();
     let lines_stream = LinesStream::new(lines);
     let fp = lines_stream.and_then(|l| {
+        dbg!(&l);
         future::ready(
             serde_json::from_str(&l).map_err(|e| std::io::Error::new(ErrorKind::Other, e)),
         )
     });
     Ok(fp)
 }
 
+fn add_to_duplicates(duplicates: &mut HashMap<usize, usize>, id1: usize, id2: usize) {
+    if id1 < id2 {
+        duplicates.insert(id1, id2);
+    }
+}
+
 impl Service {
     async fn get_task_status(&self, task_id: &str) -> Option<TaskStatus> {
         self.tasks.read().await.get(task_id).cloned()
@@ -389,6 +419,33 @@ impl Service {
                         .unwrap())
                 }
             }
+            Ok(ResourceSpec::DuplicateCandidates {
+                domain,
+                commit,
+                threshold,
+            }) => {
+                let index_id = create_index_name(&domain, &commit);
+                // if None, then return 404
+                let hnsw = self.get_index(&index_id).await.unwrap();
+                let mut duplicates: HashMap<usize, usize> = HashMap::new();
+                let elts = hnsw.layer_len(0);
+                for i in 0..elts {
+                    let current_point = &hnsw.feature(i);
+                    let results = search(current_point, 2, &hnsw).unwrap();
+                    for result in results.iter() {
+                        if f32::from_bits(result.distance()) < threshold {
+                            add_to_duplicates(&mut duplicates, i, result.internal_id())
+                        }
+                    }
+                }
+                let mut v: Vec<(&str, &str)> = duplicates
+                    .into_iter()
+                    .map(|(i, j)| (hnsw.feature(i).id(), hnsw.feature(j).id()))
+                    .collect();
+                Ok(Response::builder()
+                    .body(serde_json::to_string(&v).unwrap().into())
+                    .unwrap())
+            }
             Ok(ResourceSpec::Similar {
                 domain,
                 commit,