030e59e
zhenyi
refactor(lib): migrate storage and security tier to central lib/config (auth, cache, db, email, jwt, payment, queue, rate_limit, storage)
030e59e
use std::marker::PhantomData;
use queue::consumer::SubscribeHandle;
use queue::typed::TypedMessage;
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use crate::error::{SearchError, SearchResult};
use crate::index::{IndexHandle, IndexSettings};
use crate::metrics;
use crate::query::{SearchRequest, SearchResponse};
pub trait Searchable: Send + Sync + 'static {
const INDEX_UID: &'static str;
const PRIMARY_KEY: &'static str;
fn index_settings() -> IndexSettings {
IndexSettings::default()
}
fn document_id(&self) -> String;
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "op", content = "payload", rename_all = "snake_case")]
pub enum SyncMessage<T> {
Upsert { document: T },
Update { document: T },
Delete { id: String },
Clear {},
}
#[derive(Debug, Clone, Default)]
pub struct SyncSummary {
pub index: String,
pub operation: &'static str,
pub documents: usize,
}
impl SyncSummary {
pub fn new(index: &str, operation: &'static str, documents: usize) -> Self {
Self {
index: index.to_string(),
operation,
documents,
}
}
}
pub struct SyncedIndex<T> {
index: IndexHandle,
_marker: PhantomData<T>,
}
impl<T> Clone for SyncedIndex<T> {
fn clone(&self) -> Self {
Self {
index: self.index.clone(),
_marker: PhantomData,
}
}
}
impl<T> SyncedIndex<T>
where
T: Searchable,
{
pub(crate) fn new(index: IndexHandle) -> Self {
Self {
index,
_marker: PhantomData,
}
}
pub fn uid(&self) -> &str {
self.index.uid()
}
pub fn handle(&self) -> &IndexHandle {
&self.index
}
pub async fn ensure_index(&self) -> SearchResult<()> {
self.index.ensure_index(&T::index_settings()).await
}
}
impl<T> SyncedIndex<T>
where
T: Searchable + Clone + Serialize + DeserializeOwned + Send + Sync,
{
pub fn bridge<'a>(&self, queue: &'a queue::Queue) -> SyncBridge<'a, T> {
SyncBridge::new(queue, self.clone())
}
}
impl<T> SyncedIndex<T>
where
T: Searchable + Serialize,
{
pub async fn upsert(&self, documents: &[T]) -> SearchResult<SyncSummary> {
let _timer = metrics::Timer::start(metrics::OP_DOCS_ADD);
let batch = self.index.batch_size();
for chunk in documents.chunks(batch) {
self.index.add_documents(chunk).await?;
}
Ok(SyncSummary::new(
self.uid(),
metrics::OP_DOCS_ADD,
documents.len(),
))
}
pub async fn update(&self, documents: &[T]) -> SearchResult<SyncSummary> {
let _timer = metrics::Timer::start(metrics::OP_DOCS_UPDATE);
let batch = self.index.batch_size();
for chunk in documents.chunks(batch) {
self.index.update_documents(chunk).await?;
}
Ok(SyncSummary::new(
self.uid(),
metrics::OP_DOCS_UPDATE,
documents.len(),
))
}
pub async fn delete(&self, ids: &[String]) -> SearchResult<SyncSummary> {
self.index.delete_documents(ids).await?;
Ok(SyncSummary::new(
self.uid(),
metrics::OP_DOCS_DELETE,
ids.len(),
))
}
pub async fn clear(&self) -> SearchResult<SyncSummary> {
self.index.clear().await?;
Ok(SyncSummary::new(self.uid(), metrics::OP_DOCS_CLEAR, 0))
}
}
impl<T> SyncedIndex<T>
where
T: Searchable + DeserializeOwned,
{
pub async fn search(&self, request: &SearchRequest) -> SearchResult<SearchResponse<T>> {
self.index.search(request).await
}
}
pub struct SyncBridge<'a, T> {
queue: &'a queue::Queue,
index: SyncedIndex<T>,
}
impl<'a, T> SyncBridge<'a, T>
where
T: Searchable + Clone + Serialize + DeserializeOwned + Send + Sync,
{
pub(crate) fn new(queue: &'a queue::Queue, index: SyncedIndex<T>) -> Self {
Self { queue, index }
}
pub fn index(&self) -> &SyncedIndex<T> {
&self.index
}
pub async fn ensure_index(&self) -> SearchResult<()> {
self.index.ensure_index().await
}
pub async fn publish_upsert(&self, document: &T) -> SearchResult<()> {
self.publish(&SyncMessage::Upsert {
document: document.clone(),
})
.await
}
pub async fn publish_update(&self, document: &T) -> SearchResult<()> {
self.publish(&SyncMessage::Update {
document: document.clone(),
})
.await
}
pub async fn publish_delete(&self, id: impl Into<String>) -> SearchResult<()> {
self.publish(&SyncMessage::Delete { id: id.into() }).await
}
pub async fn publish_clear(&self) -> SearchResult<()> {
self.publish(&SyncMessage::Clear {}).await
}
async fn publish(&self, message: &SyncMessage<T>) -> SearchResult<()> {
self.queue
.typed::<SyncMessage<T>>()
.publish(message)
.await
.map_err(SearchError::Queue)
}
pub async fn spawn_consumer(
&self,
mode: queue::SubscriptionMode,
) -> Result<SubscribeHandle, SearchError> {
self.index.ensure_index().await?;
let index = self.index.clone();
let handle = self
.queue
.typed::<SyncMessage<T>>()
.subscribe(mode, move |msg: TypedMessage<SyncMessage<T>>| {
let index = index.clone();
async move {
let result = match msg.value {
SyncMessage::Upsert { document } => index.upsert(&[document]).await,
SyncMessage::Update { document } => index.update(&[document]).await,
SyncMessage::Delete { id } => index.delete(&[id]).await,
SyncMessage::Clear {} => index.clear().await,
};
match result {
Ok(summary) => {
tracing::debug!(
index = %summary.index,
operation = summary.operation,
documents = summary.documents,
"search sync applied"
);
Ok(())
}
Err(e) => {
tracing::warn!(
index = %index.uid(),
topic = %msg.meta.topic,
error = %e,
"search sync message failed"
);
Err(queue::QueueError::Handler(e.to_string()))
}
}
}
})
.map_err(SearchError::Queue)?;
Ok(handle)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::client::SearchClient;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
struct Dummy {
id: String,
name: String,
}
impl Searchable for Dummy {
const INDEX_UID: &'static str = "dummy";
const PRIMARY_KEY: &'static str = "id";
fn document_id(&self) -> String {
self.id.clone()
}
}
#[test]
fn sync_message_roundtrip() {
let msg = SyncMessage::Upsert {
document: Dummy {
id: "1".into(),
name: "x".into(),
},
};
let json = serde_json::to_string(&msg).unwrap();
assert!(json.contains("\"op\":\"upsert\""));
let back: SyncMessage<Dummy> = serde_json::from_str(&json).unwrap();
assert_eq!(msg, back);
}
#[test]
fn delete_message_roundtrip() {
let msg = SyncMessage::<Dummy>::Delete { id: "42".into() };
let json = serde_json::to_string(&msg).unwrap();
assert!(json.contains("\"op\":\"delete\""));
let back: SyncMessage<Dummy> = serde_json::from_str(&json).unwrap();
assert_eq!(msg, back);
}
#[test]
fn synced_uid_respects_prefix() {
let client = SearchClient::new("http://127.0.0.1:7700", None::<String>).unwrap();
let synced = client.synced::<Dummy>();
assert_eq!(synced.uid(), "untitled.dummy");
}
}