Skip to main content

p_memory/
notes.rs

1use crate::{storage::{self, KnowledgeBase}, text, types::*, Error, Result};
2use rusqlite::{params, Connection, OptionalExtension};
3use serde::{Deserialize, Serialize};
4use serde_json::json;
5use std::collections::{BTreeMap, BTreeSet, HashMap};
6use std::path::PathBuf;
7
8fn default_chunk_chars() -> usize { 220 }
9
10/// 按文件路径同步一篇笔记的入参:库自己读文件,标题取文件名(去扩展名)。
11/// 路径存为笔记自己的一列:领域登记过根目录时存减掉根目录的相对路径,否则逐字符原样存。
12/// 它既用来读回正文,也用来定位同一文件。
13/// 正文真相源是文件:清洗只作用于送进索引的文本,库内不留任何正文副本。
14#[derive(Debug, Clone, Serialize, Deserialize)]
15pub struct NoteFileInput {
16    #[serde(flatten)] pub record: RecordInput,
17    /// 要读取的文件路径,同时作为 `(namespace, scope, path)` 的定位键。
18    pub path: PathBuf,
19    #[serde(default = "default_chunk_chars")] pub chunk_chars: usize,
20}
21impl NoteFileInput {
22    pub fn new(path: impl Into<PathBuf>) -> Self {
23        Self { record: RecordInput::default(), path: path.into(), chunk_chars: default_chunk_chars() }
24    }
25}
26
27/// 笔记的对外形态:库内只留路径,正文读时按 `source` 读文件,标题由文件名派生。
28#[derive(Debug, Clone, Serialize, Deserialize)]
29pub struct Note {
30    #[serde(flatten)] pub header: RecordHeader,
31    /// 文件路径:写入时给进来的那条;库内一行存的是相对根目录的形态,读回时拼成绝对路径。
32    pub source: String,
33    /// 由 `source` 的文件名派生,不落库。
34    #[serde(default)] pub title: String,
35    pub chunk_chars: usize,
36}
37/// 切片不再重复携带 source/title;路径与标题经 `note_id` 关联 notes 取回。
38/// 切片正文只在索引里存一份:写入时切好、就地写进索引,`content` 读取时从索引取回。
39#[derive(Debug, Clone, Serialize, Deserialize)]
40pub struct Chunk {
41    #[serde(flatten)] pub header: RecordHeader,
42    pub note_id: i64, pub ordinal: usize, pub offset: usize, pub limit: usize,
43    #[serde(default)] pub content: String,
44}
45/// `offset` 为 1 起始的起始行,`limit` 为行数。切片正文由 `content` 自带,
46/// 超长段落被多个切片共享同一行号也不影响取回。
47#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
48pub struct TextChunk { pub ordinal: usize, pub offset: usize, pub limit: usize, pub content: String }
49
50/// Paragraph-aware chunking. Fenced code and tables are atomic, even above the
51/// target size. A long prose line may span chunks sharing that same line number.
52pub fn chunk_text(content: &str, target: usize) -> Result<Vec<TextChunk>> {
53    if !(16..=100_000).contains(&target) { return Err(Error::Validation("chunk_chars must be between 16 and 100000".into())); }
54    let lines: Vec<&str> = content.lines().collect();
55    let mut blocks: Vec<(usize, usize, String, bool)> = Vec::new();
56    let mut i = 0;
57    while i < lines.len() {
58        if lines[i].trim().is_empty() { i += 1; continue; }
59        let start = i;
60        let trimmed = lines[i].trim_start();
61        let fence_char = trimmed.chars().next().filter(|c| *c == '`' || *c == '~');
62        let fence_len = fence_char.map(|c| trimmed.chars().take_while(|x| *x == c).count()).unwrap_or(0);
63        let is_fence = fence_len >= 3;
64        let is_table = lines[i].contains('|') && i + 1 < lines.len() && {
65            let next = lines[i + 1].trim();
66            next.contains('-') && next.contains('|') && next.chars().all(|c| matches!(c, '-' | ':' | '|' | ' ' | '\t'))
67        };
68        i += 1;
69        if is_fence {
70            while i < lines.len() {
71                let line = lines[i].trim();
72                i += 1;
73                if line.chars().take_while(|c| Some(*c) == fence_char).count() >= fence_len
74                    && line.chars().all(|c| Some(c) == fence_char || c.is_whitespace()) { break; }
75            }
76        } else if is_table {
77            while i < lines.len() && lines[i].contains('|') && !lines[i].trim().is_empty() { i += 1; }
78        } else {
79            while i < lines.len() && !lines[i].trim().is_empty() {
80                let line = lines[i].trim_start();
81                if line.starts_with("```") || line.starts_with("~~~") || line.starts_with('#') { break; }
82                if i + 1 < lines.len() && lines[i].contains('|') && lines[i + 1].contains("---") { break; }
83                i += 1;
84            }
85        }
86        blocks.push((start, i, lines[start..i].join("\n"), is_fence || is_table));
87    }
88    let mut chunks = Vec::new();
89    for (start, end, body, atomic) in blocks {
90        if atomic || body.chars().count() <= target {
91            chunks.push(TextChunk { ordinal: 0, offset: start + 1, limit: end - start, content: body });
92            continue;
93        }
94        let mut current = String::new();
95        let mut count = 0;
96        let mut first_line = start + 1;
97        let mut last_line = first_line;
98        for (line_no, line) in lines.iter().enumerate().take(end).skip(start) {
99            if !current.is_empty() {
100                if count + 1 >= target {
101                    chunks.push(TextChunk { ordinal: 0, offset: first_line, limit: last_line - first_line + 1, content: std::mem::take(&mut current) });
102                    count = 0;
103                } else { current.push('\n'); count += 1; }
104            }
105            for ch in line.chars() {
106                if count == target {
107                    chunks.push(TextChunk { ordinal: 0, offset: first_line, limit: last_line - first_line + 1, content: std::mem::take(&mut current) });
108                    count = 0;
109                }
110                if current.is_empty() { first_line = line_no + 1; }
111                current.push(ch); count += 1; last_line = line_no + 1;
112            }
113        }
114        if !current.is_empty() { chunks.push(TextChunk { ordinal: 0, offset: first_line, limit: last_line - first_line + 1, content: current }); }
115    }
116    for (ordinal, chunk) in chunks.iter_mut().enumerate() { chunk.ordinal = ordinal; }
117    Ok(chunks)
118}
119
120pub(crate) fn sync_file(conn: &Connection, input: &NoteFileInput) -> Result<(Note, Vec<crate::index::IndexDocument>)> {
121    let content = std::fs::read_to_string(&input.path)?;
122    let given = input.path.to_string_lossy().into_owned();
123    let title = input.path.file_stem().map(|stem| stem.to_string_lossy().into_owned()).unwrap_or_default();
124    storage::validate_identity("source", &given)?;
125    let split = chunk_text(&content, input.chunk_chars)?;
126    let mut record = input.record.clone();
127    let namespace_id = storage::term_id(conn, &record.namespace)?;
128    let scope_id = storage::term_id(conn, &record.scope)?;
129    // 领域必须登记根目录:写入路径与它比对前缀,不重合直接拒绝、重合的部分裁掉。
130    // 库里存的永远是相对路径——项目从头到尾不知道前面的绝对路径是什么。
131    let root = storage::namespace_root(conn, namespace_id)?.ok_or_else(|| Error::Validation(
132        format!("namespace {} has no registered domain root; call notes.set_root before upserting notes", record.namespace)))?;
133    let source = relative_note_path(&root, &given)?;
134    let (dirs, split_stem) = storage::split_note_path(&source);
135    // 文件名以写入时取好的 `title`(`file_stem`)为准,与落库的 `notes.name` 同源;
136    // `split_note_path` 只用来取目录段。
137    let stem = if title.is_empty() { split_stem } else { title.clone() };
138    // 这一份标签挂到这篇的每一条切片上:调用方给的 + 路径拆出来的。
139    let mut merged = record.tags.clone();
140    merged.extend(dirs.iter().cloned());
141    if !stem.is_empty() { merged.push(stem.clone()); }
142    record.tags = merged;
143    let existing: Option<i64> = conn.query_row("SELECT record_id FROM notes WHERE namespace_id=?1 AND scope_id=?2 AND path=?3",
144        params![namespace_id, scope_id, source], |r| r.get(0)).optional()?;
145    if let Some(id) = existing {
146        if record.id.is_some_and(|given| given != id) { return Err(Error::Conflict("source already belongs to another note ID".into())); }
147        record.id = Some(id);
148    }
149    // 笔记这条记录自己不承载正文,也不占索引文档:路径信息以标签形态挂在它的每个切片上,
150    // 要文件列表就按库里的标签翻笔记。
151    let (header, _) = storage::put_record(conn, RecordKind::Note, &record,
152        &json!({"chunk_chars":input.chunk_chars}), "")?;
153    let mut documents = Vec::new();
154    // 文件名在写入这一刻就从路径取好(`file_stem` 认平台分隔符),随笔记落库:
155    // 索引那一列直接读它,重建时也不必再拆一次路径。
156    conn.execute("INSERT INTO notes(record_id,namespace_id,scope_id,path,name) VALUES (?1,?2,?3,?4,?5)
157        ON CONFLICT(record_id) DO UPDATE SET namespace_id=excluded.namespace_id,scope_id=excluded.scope_id,path=excluded.path,name=excluded.name",
158        params![header.id, namespace_id, scope_id, source, title])?;
159    // Reuse the record ID of any slice whose ordinal和内容都未变,让它的向量继续有效。
160    let mut old = Vec::new();
161    {
162        let mut stmt = conn.prepare("SELECT record_id,ordinal,fingerprint FROM chunks WHERE note_id=?1")?;
163        for row in stmt.query_map([header.id], |r| Ok((r.get::<_, i64>(0)?, r.get::<_, i64>(1)?, r.get::<_, String>(2)?)))? {
164            old.push(row?);
165        }
166    }
167    let reuse: HashMap<(i64, String), i64> = old.iter().map(|(id, ordinal, fp)| ((*ordinal, fp.clone()), *id)).collect();
168    // Ordinals may be reshuffled between revisions, so clear the projection first
169    // and rebuild it; slices whose ordinal and content are unchanged keep their
170    // record ID (and therefore their embedding).
171    conn.execute("DELETE FROM chunks WHERE note_id=?1", [header.id])?;
172    let mut used = BTreeSet::new();
173    for chunk in &split {
174        let content_digest = text::digest(&chunk.content);
175        let payload = json!({"note_id":header.id,"ordinal":chunk.ordinal,"offset":chunk.offset,"limit":chunk.limit});
176        let (id, document) = match reuse.get(&(chunk.ordinal as i64, content_digest.clone())) {
177            Some(&id) => {
178                used.insert(id);
179                conn.execute("UPDATE records SET payload_json=?2 WHERE id=?1", params![id, serde_json::to_string(&payload)?])?;
180                // 内容没变,但标签是这篇当前这一份:标签换了指纹跟着换,旧向量就地作废。
181                let tag_ids = storage::set_record_tags(conn, id, &header.tags)?;
182                let fingerprint = storage::record_fingerprint(&chunk.content, &header.tags);
183                conn.execute("UPDATE records SET fingerprint=?2,updated_at_us=MAX(updated_at_us,?3) WHERE id=?1",
184                    params![id, fingerprint, header.updated_at_us])?;
185                conn.execute("DELETE FROM embeddings WHERE record_id=?1 AND fingerprint<>?2", params![id, fingerprint])?;
186                // 标签与指纹都动过,这个领域的向量分区跟着变。
187                storage::touch_namespace(&header.namespace);
188                let (name, path, exclude) = storage::index_columns(conn, RecordKind::Chunk, &payload);
189                (id, crate::index::IndexDocument { id, namespace_id, scope_id, kind: RecordKind::Chunk,
190                    text: chunk.content.clone(),
191                    name, path,
192                    note_id: header.id,
193                    tags_prefix: storage::tags_prefix(RecordKind::Chunk, &header.tags, &exclude, &payload),
194                    tag_ids })
195            }
196            None => {
197                let chunk_input = RecordInput { id: None, namespace: header.namespace.clone(), scope: header.scope.clone(), tags: header.tags.clone(),
198                    evidence: vec![], metadata: header.metadata.clone(), created_at_us: Some(header.created_at_us), updated_at_us: Some(header.updated_at_us), expected_revision: None };
199                let (chunk_header, document) = storage::put_record(conn, RecordKind::Chunk, &chunk_input, &payload, &chunk.content)?;
200                (chunk_header.id, document)
201            }
202        };
203        conn.execute("INSERT INTO chunks(record_id,note_id,ordinal,\"offset\",\"limit\",fingerprint) VALUES (?1,?2,?3,?4,?5,?6)",
204            params![id, header.id, chunk.ordinal as i64, chunk.offset as i64, chunk.limit as i64, content_digest])?;
205        documents.push(document);
206    }
207    // Delete obsolete records (and their embeddings), not just projection rows.
208    for (id, _, _) in old { if !used.contains(&id) { storage::delete_record(conn, &RecordKey { id })?; } }
209    Ok((Note { header, source: given, title, chunk_chars: input.chunk_chars }, documents))
210}
211
212/// 减掉领域根目录得到库里存的那条相对路径;不在根目录之内直接报错,不做猜测。
213/// 只统一分隔符,不改大小写——存的是什么名字,读回文件时就找什么名字。
214fn relative_note_path(root: &str, given: &str) -> Result<String> {
215    let root = root.replace('\\', "/");
216    let root = root.trim_end_matches('/');
217    let relative = given.replace('\\', "/");
218    let Some(relative) = relative.strip_prefix(root).and_then(|rest| rest.strip_prefix('/')) else {
219        return Err(Error::Validation(format!("note path {given} is outside the domain root {root}")));
220    };
221    if relative.is_empty() { return Err(Error::Validation("note path must name a file below the domain root".into())); }
222    Ok(relative.to_string())
223}
224
225#[derive(Clone)]
226pub struct NoteStore(pub(crate) KnowledgeBase);
227
228impl NoteStore {
229    /// 从索引取一批记录的正文。有取不到的记录说明索引还没追上这批写入,就提交一次再取;
230    /// 索引已追平时这条读路径一次都不碰写锁。
231    fn bodies(&self, conn: &Connection, ids: &[i64]) -> Result<BTreeMap<i64, String>> {
232        if ids.is_empty() { return Ok(BTreeMap::new()); }
233        let mut bodies = self.0.index()?.bodies(ids)?;
234        if ids.iter().any(|id| !bodies.contains_key(id)) {
235            self.0.sync_index_if_behind(conn)?;
236            bodies = self.0.index()?.bodies(ids)?;
237        }
238        Ok(bodies)
239    }
240    /// 读文件后同步一篇笔记:正文取文件原文,标题取文件名(去扩展名),路径即身份。
241    /// 监听与对账在使用方;库只按给定路径处理这一个文件。
242    pub fn upsert_file(&self, input: NoteFileInput) -> Result<WriteReceipt<Note>> {
243        let receipt = self.0.mutate(|tx| sync_file(tx, &input))?;
244        let WriteReceipt { value: (note, documents), revision } = receipt;
245        // 切片正文在写入时就地切好、一路带到索引,索引阶段不再回读文件。
246        self.0.index_documents(&documents)?;
247        Ok(WriteReceipt { value: note, revision })
248    }
249    /// 登记该知识领域的笔记根目录。登记之后写入的笔记路径必须是它的子路径:
250    /// 库里存相对路径,相对路径按段拆出的标签挂到这篇的每一条切片上。
251    pub fn set_root(&self, namespace: &str, root: &str) -> Result<()> {
252        storage::validate_identity("namespace", namespace)?;
253        if !std::path::Path::new(root).is_dir() {
254            return Err(Error::Validation(format!("domain root {root} is not an existing directory")));
255        }
256        let root = root.replace('\\', "/");
257        let root = root.trim_end_matches('/').to_string();
258        self.0.write(|writer| {
259            let namespace_id = storage::term_id(&writer.conn, namespace)?;
260            writer.conn.execute("INSERT INTO namespace_roots(namespace_id,root) VALUES (?1,?2)
261                ON CONFLICT(namespace_id) DO UPDATE SET root=excluded.root", params![namespace_id, root])?;
262            Ok(())
263        })
264    }
265    /// 该领域登记的笔记根目录;没登记就是 `None`。
266    pub fn root(&self, namespace: &str) -> Result<Option<String>> {
267        let state = self.0.read()?;
268        let conn = state.conn();
269        let namespace_id: Option<i64> = conn.query_row("SELECT id FROM strings WHERE text=?1",
270            [text::normalized_tag(namespace)], |r| r.get(0)).optional()?;
271        let Some(namespace_id) = namespace_id else { return Ok(None) };
272        storage::namespace_root(conn, namespace_id)
273    }
274    pub fn get(&self, id: i64, filter: &ReadFilter) -> Result<Note> {
275        storage::get(self.0.read()?.conn(), &RecordKey { id }, filter)
276    }
277    pub fn list(&self, page: &PageRequest) -> Result<Page<Note>> { storage::list(self.0.read()?.conn(), RecordKind::Note, page) }
278    pub fn get_chunk(&self, id: i64, filter: &ReadFilter) -> Result<Chunk> {
279        let state = self.0.read()?;
280        let conn = state.conn();
281        let mut chunk: Chunk = storage::get(conn, &RecordKey { id }, filter)?;
282        chunk.content = self.bodies(conn, &[id])?.remove(&id).unwrap_or_default();
283        Ok(chunk)
284    }
285    pub fn chunks(&self, note_id: i64, filter: &ReadFilter) -> Result<Vec<Chunk>> {
286        let state = self.0.read()?;
287        let conn = state.conn();
288        let _: Note = storage::get(conn, &RecordKey { id: note_id }, filter)?;
289        let ids: Vec<i64> = {
290            let mut stmt = conn.prepare("SELECT record_id FROM chunks WHERE note_id=?1 ORDER BY ordinal")?;
291            let rows = stmt.query_map([note_id], |r| r.get::<_, i64>(0))?;
292            rows.collect::<std::result::Result<Vec<_>, _>>()?
293        };
294        // 一次批量取回:逐条 `get` 会为每片各跑一遍过滤与装配,一篇几百片就是几百次回库。
295        let mut loaded: BTreeMap<i64, Chunk> = storage::load_many(conn, &ids, filter)?;
296        let mut chunks: Vec<Chunk> = ids.iter().filter_map(|id| loaded.remove(id)).collect();
297        // 一批切片一次取回正文:正文只存在索引里,逐条回库或回源文件都没有意义。
298        let ids: Vec<i64> = chunks.iter().map(|chunk| chunk.header.id).collect();
299        let mut bodies = self.bodies(conn, &ids)?;
300        for chunk in &mut chunks { chunk.content = bodies.remove(&chunk.header.id).unwrap_or_default(); }
301        Ok(chunks)
302    }
303    pub fn delete(&self, id: i64, filter: &ReadFilter) -> Result<WriteReceipt<bool>> {
304        self.0.mutate(|tx| delete_note(tx, id, filter))
305    }
306    /// 删除过滤条件命中的全部笔记,返回删除条数。空命中返回 0。
307    /// 每篇都连同它的切片一起删:`chunks.note_id` 是 RESTRICT 引用,先切片后笔记。
308    pub fn delete_by_filter(&self, filter: &ReadFilter) -> Result<WriteReceipt<usize>> {
309        self.0.mutate(|tx| {
310            let mut removed = 0;
311            for id in storage::select_ids(tx, filter, &[RecordKind::Note])? {
312                if delete_note(tx, id, filter)? { removed += 1; }
313            }
314            Ok(removed)
315        })
316    }
317}
318
319/// 删一篇笔记:先删它的切片,再删笔记本身。
320fn delete_note(conn: &Connection, id: i64, filter: &ReadFilter) -> Result<bool> {
321    let key = RecordKey { id };
322    let _: Note = storage::get(conn, &key, filter)?;
323    let mut stmt = conn.prepare("SELECT record_id FROM chunks WHERE note_id=?1")?;
324    let ids = stmt.query_map([id], |r| r.get::<_, i64>(0))?.collect::<std::result::Result<Vec<_>, _>>()?;
325    for child in ids { storage::delete_record(conn, &RecordKey { id: child })?; }
326    storage::delete_record(conn, &key)
327}
328
329#[cfg(test)]
330mod tests {
331    use super::*;
332
333    #[test]
334    fn split_note_path_separates_directories_from_the_file_name() {
335        assert_eq!(storage::split_note_path("notes/characters/overview.md"), (vec!["notes".to_string(), "characters".to_string()], "overview".to_string()));
336        // 目录段里的点不是扩展名:只有最后一段去后缀。
337        assert_eq!(storage::split_note_path("v1.2/角色.设定.md"), (vec!["v1.2".to_string()], "角色.设定".to_string()));
338        assert_eq!(storage::split_note_path("overview"), (Vec::new(), "overview".to_string()));
339    }
340
341    #[test]
342    fn note_paths_must_stay_inside_the_domain_root() {
343        assert_eq!(relative_note_path("E:/data/demo", r"E:\data\demo\notes\overview.md").unwrap(), "notes/overview.md");
344        // 前缀只是像,不是子路径;也不许正好等于根目录本身。
345        assert!(relative_note_path("E:/data/demo", "E:/data/demox/overview.md").is_err());
346        assert!(relative_note_path("E:/data/demo", "E:/data/other/overview.md").is_err());
347        assert!(relative_note_path("E:/data/demo", "E:/data/demo").is_err());
348    }
349}