1use super::types::{EdgeProps, EdgeRef, IndexCreateRequest};
3use super::*;
4use crate::outbox::OutboxPayload;
5use crate::types::ChangeType;
6use linkrs_core::types::TransactionId;
7use linkrs_core::Value;
8#[cfg_attr(
9 not(any(feature = "fulltext", feature = "vector")),
10 allow(unused_variables)
11)]
12impl super::SyncManager {
13 pub fn on_vertex_change_with_txn(
14 &self,
15 txn_id: TransactionId,
16 space_id: u64,
17 tag_name: &str,
18 vertex_id: &Value,
19 properties: &[(Arc<str>, Value)],
20 change_type: ChangeType,
21 ) -> Result<(), SyncError> {
22 let staged = self.expand_auto_embed_text(space_id, tag_name, properties, change_type)?;
23 self.stage_intent(
24 txn_id,
25 OutboxPayload::Vertex {
26 space_id,
27 tag_name: tag_name.to_string(),
28 vertex_id: vertex_id.clone(),
29 properties: staged,
30 change_type,
31 },
32 )?;
33 Ok(())
34 }
35
36 #[cfg(not(feature = "vector"))]
37 fn expand_auto_embed_text(
38 &self,
39 _space_id: u64,
40 _tag_name: &str,
41 properties: &[(Arc<str>, Value)],
42 _change_type: ChangeType,
43 ) -> Result<Vec<(Arc<str>, Value)>, SyncError> {
44 Ok(properties.to_vec())
45 }
46
47 #[cfg(feature = "vector")]
48 fn expand_auto_embed_text(
49 &self,
50 space_id: u64,
51 tag_name: &str,
52 properties: &[(Arc<str>, Value)],
53 change_type: ChangeType,
54 ) -> Result<Vec<(Arc<str>, Value)>, SyncError> {
55 if !self.auto_embed_text() {
56 return Ok(properties.to_vec());
57 }
58 if matches!(change_type, ChangeType::Delete) {
59 return Ok(properties.to_vec());
60 }
61 let Some(coordinator) = self.vector_coordinator.as_ref() else {
62 return Err(SyncError::VectorError(
63 "auto_embed_text requires a configured vector coordinator".to_string(),
64 ));
65 };
66 use std::collections::HashSet;
67 let explicit_vectors: HashSet<&str> = properties
68 .iter()
69 .filter(|(_, value)| value.as_vector().is_some())
70 .map(|(field, _)| field.as_ref())
71 .collect();
72 let mut pending_fields = Vec::new();
73 let mut pending_texts = Vec::new();
74 for (field, value) in properties {
75 if explicit_vectors.contains(field.as_ref()) {
76 continue;
77 }
78 let Some(text) = value.string_value() else {
79 continue;
80 };
81 if !coordinator.index_exists(space_id, tag_name, field) {
82 continue;
83 }
84 pending_fields.push(field.clone());
85 pending_texts.push(text);
86 }
87 if pending_fields.is_empty() {
88 return Ok(properties.to_vec());
89 }
90 #[cfg(not(feature = "embedding"))]
91 {
92 Err(SyncError::VectorError(
93 "auto_embed_text requires the embedding feature and service".to_string(),
94 ))
95 }
96 #[cfg(feature = "embedding")]
97 {
98 if coordinator.embedding_service().is_none() {
99 return Err(SyncError::VectorError(
100 "auto_embed_text requires a configured embedding service".to_string(),
101 ));
102 }
103 let vectors = crate::runtime::block_on_ambient(coordinator.embed_texts(&pending_texts))
104 .map_err(|error| SyncError::VectorError(error.to_string()))?
105 .map_err(|error| SyncError::VectorError(error.to_string()))?;
106 if vectors.len() != pending_fields.len() {
107 return Err(SyncError::VectorError(
108 "embedding service returned mismatched vector count".to_string(),
109 ));
110 }
111 let mut staged = properties.to_vec();
112 for (field, vector) in pending_fields.into_iter().zip(vectors) {
113 if let Some(info) = coordinator.index_info(space_id, tag_name, &field) {
114 let expected = info.config.vector_size;
115 if expected != 0 && vector.len() != expected {
116 return Err(SyncError::VectorError(format!(
117 "auto_embed dimension mismatch for {}.{}.{}: expected {} got {}",
118 space_id,
119 tag_name,
120 field,
121 expected,
122 vector.len()
123 )));
124 }
125 }
126 staged.push((field, Value::vector(vector)));
127 }
128 Ok(staged)
129 }
130 }
131
132 pub fn on_index_create(
133 &self,
134 txn_id: TransactionId,
135 request: IndexCreateRequest,
136 ) -> Result<(), SyncError> {
137 self.stage_intent(
138 txn_id,
139 OutboxPayload::CreateIndex {
140 space_id: request.space_id,
141 index_name: request.index_name,
142 schema_name: request.schema_name,
143 index_type: request.index_type,
144 fields: request.fields,
145 properties: request.properties,
146 },
147 )?;
148 Ok(())
149 }
150
151 pub fn on_index_drop(
152 &self,
153 txn_id: TransactionId,
154 space_id: u64,
155 index_name: &str,
156 schema_name: &str,
157 index_type: &str,
158 fields: &[String],
159 ) -> Result<(), SyncError> {
160 self.stage_intent(
161 txn_id,
162 OutboxPayload::DropIndex {
163 space_id,
164 index_name: index_name.to_string(),
165 schema_name: schema_name.to_string(),
166 index_type: index_type.to_string(),
167 fields: fields.to_vec(),
168 },
169 )?;
170 Ok(())
171 }
172
173 pub fn on_space_drop(&self, txn_id: TransactionId, space_id: u64) -> Result<(), SyncError> {
174 self.stage_intent(txn_id, OutboxPayload::DropSpace { space_id })?;
175 Ok(())
176 }
177
178 pub fn on_tag_drop(
179 &self,
180 txn_id: TransactionId,
181 space_id: u64,
182 tag_name: &str,
183 ) -> Result<(), SyncError> {
184 self.stage_intent(
185 txn_id,
186 OutboxPayload::DropTag {
187 space_id,
188 tag_name: tag_name.to_string(),
189 },
190 )?;
191 Ok(())
192 }
193
194 pub fn on_edge_insert(
195 &self,
196 txn_id: TransactionId,
197 space_id: u64,
198 edge: &linkrs_core::Edge,
199 ) -> Result<(), SyncError> {
200 self.stage_intent(
201 txn_id,
202 OutboxPayload::EdgeInsert {
203 space_id,
204 edge: edge.clone(),
205 },
206 )?;
207 Ok(())
208 }
209
210 pub fn on_edge_delete(
211 &self,
212 txn_id: TransactionId,
213 space_id: u64,
214 src: &Value,
215 dst: &Value,
216 edge_type: &str,
217 ranking: i64,
218 ) -> Result<(), SyncError> {
219 self.stage_intent(
220 txn_id,
221 OutboxPayload::EdgeDelete {
222 space_id,
223 src: src.clone(),
224 dst: dst.clone(),
225 edge_type: edge_type.to_string(),
226 ranking,
227 },
228 )?;
229 Ok(())
230 }
231
232 pub fn on_edge_update(
233 &self,
234 _txn_id: TransactionId,
235 _space_id: u64,
236 _edge: EdgeRef<'_>,
237 _props: EdgeProps<'_>,
238 ) -> Result<(), SyncError> {
239 Ok(())
240 }
241}