356 lines
11 KiB
Rust
356 lines
11 KiB
Rust
|
use event_log::DomainEventRecord;
|
||
|
|
|
||
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
||
|
|
pub enum IndexedEntityKind {
|
||
|
|
PageTitle,
|
||
|
|
PageSummary,
|
||
|
|
BlockContent,
|
||
|
|
BlockPath,
|
||
|
|
}
|
||
|
|
|
||
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
||
|
|
pub struct IndexedDocument {
|
||
|
|
pub workspace_id: String,
|
||
|
|
pub entity_kind: IndexedEntityKind,
|
||
|
|
pub entity_id: String,
|
||
|
|
pub parent_id: Option<String>,
|
||
|
|
pub title: Option<String>,
|
||
|
|
pub content: String,
|
||
|
|
pub updated_at: String,
|
||
|
|
pub source_event_id: Option<String>,
|
||
|
|
pub revision: Option<u64>,
|
||
|
|
}
|
||
|
|
|
||
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
||
|
|
pub struct IndexCursor {
|
||
|
|
pub workspace_id: String,
|
||
|
|
pub last_processed_event_id: String,
|
||
|
|
pub last_processed_at: String,
|
||
|
|
}
|
||
|
|
|
||
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
||
|
|
pub struct SearchHit {
|
||
|
|
pub workspace_id: String,
|
||
|
|
pub entity_kind: IndexedEntityKind,
|
||
|
|
pub entity_id: String,
|
||
|
|
pub snippet: Option<String>,
|
||
|
|
}
|
||
|
|
|
||
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
||
|
|
pub struct SearchResultSet {
|
||
|
|
pub query: String,
|
||
|
|
pub workspace_id: Option<String>,
|
||
|
|
pub page_id: Option<String>,
|
||
|
|
pub hits: Vec<SearchHit>,
|
||
|
|
}
|
||
|
|
|
||
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
||
|
|
pub struct ProjectedDocumentBatch {
|
||
|
|
pub workspace_id: String,
|
||
|
|
pub source_event_id: String,
|
||
|
|
pub documents: Vec<IndexedDocument>,
|
||
|
|
}
|
||
|
|
|
||
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
||
|
|
pub struct ProjectionResult {
|
||
|
|
pub cursor: IndexCursor,
|
||
|
|
pub batches: Vec<ProjectedDocumentBatch>,
|
||
|
|
}
|
||
|
|
|
||
|
|
pub trait DomainEventProjector {
|
||
|
|
fn project(&self, event: &DomainEventRecord) -> Vec<IndexedDocument>;
|
||
|
|
}
|
||
|
|
|
||
|
|
pub fn supported_index_objects() -> Vec<IndexedEntityKind> {
|
||
|
|
vec![
|
||
|
|
IndexedEntityKind::PageTitle,
|
||
|
|
IndexedEntityKind::PageSummary,
|
||
|
|
IndexedEntityKind::BlockContent,
|
||
|
|
IndexedEntityKind::BlockPath,
|
||
|
|
]
|
||
|
|
}
|
||
|
|
|
||
|
|
pub fn can_rebuild_from_events(cursor: &IndexCursor) -> bool {
|
||
|
|
!cursor.workspace_id.trim().is_empty() && !cursor.last_processed_event_id.trim().is_empty()
|
||
|
|
}
|
||
|
|
|
||
|
|
pub fn advance_workspace_cursor(
|
||
|
|
cursor: &IndexCursor,
|
||
|
|
event_id: impl Into<String>,
|
||
|
|
processed_at: impl Into<String>,
|
||
|
|
) -> IndexCursor {
|
||
|
|
IndexCursor {
|
||
|
|
workspace_id: cursor.workspace_id.clone(),
|
||
|
|
last_processed_event_id: event_id.into(),
|
||
|
|
last_processed_at: processed_at.into(),
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
pub fn project_domain_event<P: DomainEventProjector>(
|
||
|
|
projector: &P,
|
||
|
|
event: &DomainEventRecord,
|
||
|
|
cursor: &IndexCursor,
|
||
|
|
) -> ProjectionResult {
|
||
|
|
let documents = projector.project(event);
|
||
|
|
ProjectionResult {
|
||
|
|
cursor: advance_workspace_cursor(
|
||
|
|
cursor,
|
||
|
|
event.event_id.clone(),
|
||
|
|
event.created_at.as_str().to_string(),
|
||
|
|
),
|
||
|
|
batches: if documents.is_empty() {
|
||
|
|
Vec::new()
|
||
|
|
} else {
|
||
|
|
vec![ProjectedDocumentBatch {
|
||
|
|
workspace_id: event.workspace_id.clone(),
|
||
|
|
source_event_id: event.event_id.clone(),
|
||
|
|
documents,
|
||
|
|
}]
|
||
|
|
},
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
pub fn rebuild_from_events<P: DomainEventProjector>(
|
||
|
|
projector: &P,
|
||
|
|
events: &[DomainEventRecord],
|
||
|
|
cursor: &IndexCursor,
|
||
|
|
) -> ProjectionResult {
|
||
|
|
let mut next_cursor = cursor.clone();
|
||
|
|
let mut batches = Vec::new();
|
||
|
|
for event in events {
|
||
|
|
if event.workspace_id != cursor.workspace_id {
|
||
|
|
continue;
|
||
|
|
}
|
||
|
|
let projected = projector.project(event);
|
||
|
|
next_cursor = advance_workspace_cursor(
|
||
|
|
&next_cursor,
|
||
|
|
event.event_id.clone(),
|
||
|
|
event.created_at.as_str().to_string(),
|
||
|
|
);
|
||
|
|
if !projected.is_empty() {
|
||
|
|
batches.push(ProjectedDocumentBatch {
|
||
|
|
workspace_id: event.workspace_id.clone(),
|
||
|
|
source_event_id: event.event_id.clone(),
|
||
|
|
documents: projected,
|
||
|
|
});
|
||
|
|
}
|
||
|
|
}
|
||
|
|
ProjectionResult {
|
||
|
|
cursor: next_cursor,
|
||
|
|
batches,
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
pub fn search_pages(
|
||
|
|
query: &str,
|
||
|
|
workspace_id: Option<&str>,
|
||
|
|
documents: &[IndexedDocument],
|
||
|
|
limit: usize,
|
||
|
|
) -> SearchResultSet {
|
||
|
|
let query = query.trim().to_lowercase();
|
||
|
|
let hits = documents
|
||
|
|
.iter()
|
||
|
|
.filter(|document| matches!(document.entity_kind, IndexedEntityKind::PageTitle | IndexedEntityKind::PageSummary))
|
||
|
|
.filter(|document| workspace_id.map_or(true, |workspace| document.workspace_id == workspace))
|
||
|
|
.filter(|document| document.title.as_deref().unwrap_or("").to_lowercase().contains(&query)
|
||
|
|
|| document.content.to_lowercase().contains(&query))
|
||
|
|
.take(limit)
|
||
|
|
.map(|document| SearchHit {
|
||
|
|
workspace_id: document.workspace_id.clone(),
|
||
|
|
entity_kind: document.entity_kind.clone(),
|
||
|
|
entity_id: document.entity_id.clone(),
|
||
|
|
snippet: Some(document.content.chars().take(120).collect()),
|
||
|
|
})
|
||
|
|
.collect();
|
||
|
|
|
||
|
|
SearchResultSet {
|
||
|
|
query: query.into(),
|
||
|
|
workspace_id: workspace_id.map(|value| value.into()),
|
||
|
|
page_id: None,
|
||
|
|
hits,
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
pub fn search_blocks(
|
||
|
|
query: &str,
|
||
|
|
page_id: Option<&str>,
|
||
|
|
documents: &[IndexedDocument],
|
||
|
|
limit: usize,
|
||
|
|
) -> SearchResultSet {
|
||
|
|
let query = query.trim().to_lowercase();
|
||
|
|
let hits = documents
|
||
|
|
.iter()
|
||
|
|
.filter(|document| matches!(document.entity_kind, IndexedEntityKind::BlockContent | IndexedEntityKind::BlockPath))
|
||
|
|
.filter(|document| page_id.map_or(true, |page| document.parent_id.as_deref() == Some(page) || document.entity_id == page))
|
||
|
|
.filter(|document| document.content.to_lowercase().contains(&query))
|
||
|
|
.take(limit)
|
||
|
|
.map(|document| SearchHit {
|
||
|
|
workspace_id: document.workspace_id.clone(),
|
||
|
|
entity_kind: document.entity_kind.clone(),
|
||
|
|
entity_id: document.entity_id.clone(),
|
||
|
|
snippet: Some(document.content.chars().take(120).collect()),
|
||
|
|
})
|
||
|
|
.collect();
|
||
|
|
|
||
|
|
SearchResultSet {
|
||
|
|
query: query.into(),
|
||
|
|
workspace_id: None,
|
||
|
|
page_id: page_id.map(|value| value.into()),
|
||
|
|
hits,
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
pub struct MinimalWorkspaceProjector;
|
||
|
|
|
||
|
|
impl DomainEventProjector for MinimalWorkspaceProjector {
|
||
|
|
fn project(&self, event: &DomainEventRecord) -> Vec<IndexedDocument> {
|
||
|
|
let content = event.payload_json.trim();
|
||
|
|
let (entity_kind, title, parent_id) = match event.aggregate_type.as_str() {
|
||
|
|
"page" => (
|
||
|
|
IndexedEntityKind::PageTitle,
|
||
|
|
Some(event.event_type.clone()),
|
||
|
|
None,
|
||
|
|
),
|
||
|
|
"block" => (
|
||
|
|
IndexedEntityKind::BlockContent,
|
||
|
|
None,
|
||
|
|
Some(event.aggregate_id.clone()),
|
||
|
|
),
|
||
|
|
_ => return Vec::new(),
|
||
|
|
};
|
||
|
|
|
||
|
|
vec![IndexedDocument {
|
||
|
|
workspace_id: event.workspace_id.clone(),
|
||
|
|
entity_kind,
|
||
|
|
entity_id: event.aggregate_id.clone(),
|
||
|
|
parent_id,
|
||
|
|
title,
|
||
|
|
content: content.to_string(),
|
||
|
|
updated_at: event.created_at.as_str().to_string(),
|
||
|
|
source_event_id: Some(event.event_id.clone()),
|
||
|
|
revision: Some(event.event_version as u64),
|
||
|
|
}]
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
#[cfg(test)]
|
||
|
|
mod tests {
|
||
|
|
use super::*;
|
||
|
|
use core_domain::Timestamp;
|
||
|
|
use event_log::EventStatus;
|
||
|
|
|
||
|
|
fn cursor() -> IndexCursor {
|
||
|
|
IndexCursor {
|
||
|
|
workspace_id: "ws_1".into(),
|
||
|
|
last_processed_event_id: "evt_0".into(),
|
||
|
|
last_processed_at: "2026-04-11T00:00:00Z".into(),
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
fn event(
|
||
|
|
workspace_id: &str,
|
||
|
|
aggregate_type: &str,
|
||
|
|
aggregate_id: &str,
|
||
|
|
event_id: &str,
|
||
|
|
event_type: &str,
|
||
|
|
) -> DomainEventRecord {
|
||
|
|
DomainEventRecord {
|
||
|
|
event_id: event_id.into(),
|
||
|
|
workspace_id: workspace_id.into(),
|
||
|
|
aggregate_type: aggregate_type.into(),
|
||
|
|
aggregate_id: aggregate_id.into(),
|
||
|
|
event_type: event_type.into(),
|
||
|
|
event_version: 1,
|
||
|
|
payload_json: "{\"text\":\"hello\"}".into(),
|
||
|
|
command_log_id: "cmd_1".into(),
|
||
|
|
actor_type: "human".into(),
|
||
|
|
created_at: Timestamp::new("2026-04-11T00:00:01Z"),
|
||
|
|
status: EventStatus::Committed,
|
||
|
|
trace_id: "trace_1".into(),
|
||
|
|
request_id: "req_1".into(),
|
||
|
|
command_id: "cmd_1".into(),
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
#[test]
|
||
|
|
fn supports_page_and_block_projection_objects() {
|
||
|
|
assert_eq!(
|
||
|
|
supported_index_objects(),
|
||
|
|
vec![
|
||
|
|
IndexedEntityKind::PageTitle,
|
||
|
|
IndexedEntityKind::PageSummary,
|
||
|
|
IndexedEntityKind::BlockContent,
|
||
|
|
IndexedEntityKind::BlockPath
|
||
|
|
]
|
||
|
|
);
|
||
|
|
}
|
||
|
|
|
||
|
|
#[test]
|
||
|
|
fn cursor_advances_incrementally() {
|
||
|
|
let next = advance_workspace_cursor(&cursor(), "evt_2", "2026-04-11T00:00:02Z");
|
||
|
|
assert_eq!(next.last_processed_event_id, "evt_2");
|
||
|
|
assert_eq!(next.workspace_id, "ws_1");
|
||
|
|
}
|
||
|
|
|
||
|
|
#[test]
|
||
|
|
fn projector_builds_real_documents_for_page_and_block() {
|
||
|
|
let projector = MinimalWorkspaceProjector;
|
||
|
|
let page_docs =
|
||
|
|
projector.project(&event("ws_1", "page", "page_1", "evt_1", "page.created"));
|
||
|
|
let block_docs =
|
||
|
|
projector.project(&event("ws_1", "block", "block_1", "evt_2", "block.created"));
|
||
|
|
assert_eq!(page_docs.len(), 1);
|
||
|
|
assert_eq!(block_docs.len(), 1);
|
||
|
|
assert_eq!(page_docs[0].entity_kind, IndexedEntityKind::PageTitle);
|
||
|
|
assert_eq!(block_docs[0].entity_kind, IndexedEntityKind::BlockContent);
|
||
|
|
}
|
||
|
|
|
||
|
|
#[test]
|
||
|
|
fn rebuild_skips_other_workspaces_and_keeps_cursor() {
|
||
|
|
let projector = MinimalWorkspaceProjector;
|
||
|
|
let events = vec![
|
||
|
|
event("ws_2", "page", "page_x", "evt_x", "page.created"),
|
||
|
|
event("ws_1", "block", "block_1", "evt_1", "block.created"),
|
||
|
|
];
|
||
|
|
let result = rebuild_from_events(&projector, &events, &cursor());
|
||
|
|
assert_eq!(result.batches.len(), 1);
|
||
|
|
assert_eq!(result.cursor.last_processed_event_id, "evt_1");
|
||
|
|
}
|
||
|
|
|
||
|
|
#[test]
|
||
|
|
fn search_helpers_return_hits() {
|
||
|
|
let docs = vec![
|
||
|
|
IndexedDocument {
|
||
|
|
workspace_id: "ws_1".into(),
|
||
|
|
entity_kind: IndexedEntityKind::PageTitle,
|
||
|
|
entity_id: "page_1".into(),
|
||
|
|
parent_id: None,
|
||
|
|
title: Some("Rust Notes".into()),
|
||
|
|
content: "Rust notes for phase 2".into(),
|
||
|
|
updated_at: "2026-04-11T00:00:01Z".into(),
|
||
|
|
source_event_id: Some("evt_1".into()),
|
||
|
|
revision: Some(1),
|
||
|
|
},
|
||
|
|
IndexedDocument {
|
||
|
|
workspace_id: "ws_1".into(),
|
||
|
|
entity_kind: IndexedEntityKind::BlockContent,
|
||
|
|
entity_id: "block_1".into(),
|
||
|
|
parent_id: Some("page_1".into()),
|
||
|
|
title: None,
|
||
|
|
content: "A block about rust search indexing".into(),
|
||
|
|
updated_at: "2026-04-11T00:00:02Z".into(),
|
||
|
|
source_event_id: Some("evt_2".into()),
|
||
|
|
revision: Some(1),
|
||
|
|
},
|
||
|
|
];
|
||
|
|
|
||
|
|
let pages = search_pages("rust", Some("ws_1"), &docs, 10);
|
||
|
|
assert_eq!(pages.hits.len(), 1);
|
||
|
|
assert_eq!(pages.hits[0].entity_id, "page_1");
|
||
|
|
|
||
|
|
let blocks = search_blocks("search", Some("page_1"), &docs, 10);
|
||
|
|
assert_eq!(blocks.hits.len(), 1);
|
||
|
|
assert_eq!(blocks.hits[0].entity_id, "block_1");
|
||
|
|
}
|
||
|
|
}
|