Search

Introduction

Featuresearch A CRM lives or dies by its search box. Typing "ada lov" should find Ada Lovelace the contact, Lovelace Analytics the company and the "Lovelace renewal" deal, ranked by relevance, in a few milliseconds, and only the records the person is allowed to see.

navree-search adds full-text and vector search to the entities you already have. You derive Searchable on a SeaORM model, register it with SearchModule, and from then on:

• every insert, update and delete made through Repo queues a sync job in the same commit, so the index follows the database without code in your services;
• the Search service queries the index and returns your models, in relevance order, as a normal Page;
• navree search:import Deal builds or rebuilds an index from the table.

One API runs on five engines. Pick the one that fits your deployment and change it later without touching your queries:

Vector search stores embeddings in pgvector or Qdrant next to the full-text index, so one query can mix keyword and semantic relevance.

Installation

Search is not part of the default full feature set. Turn on the search feature of the navree facade:

[dependencies]
navree = { git = "https://github.com/NavreeHQ/navree", features = ["search"] }

The search feature includes the PostgreSQL, Meilisearch, Typesense and memory engines; the server engines talk HTTP through navree-http-client. The embedded engine compiles Tantivy into your binary, so it has a feature of its own:

navree = { git = "https://github.com/NavreeHQ/navree", features = ["search", "search-tantivy"] }

The postgres engine on PostgreSQL and the pgvector store need navree-db's PostgreSQL driver (the postgres feature). On SQLite the postgres engine works with the default driver.

Configuration

Add SearchModule to your app root, after the database and the queue:

use navree::prelude::*;

pub fn app(sources: ConfigSources) -> AppBuilder {
    Navree::app()
        .module(ConfigModule::for_root(sources))
        .module(DatabaseModule::from_config())
        .module(QueueModule::from_config()) // runs the sync jobs
        .module(SearchModule::from_config()) // section `search`
        .module(ContactsModule)
        .module(DealsModule)
}

The module is global: feature modules inject Arc<Search> without importing it. from_config(), and a plain .module(SearchModule), read the search section:

[search]
engine = "meilisearch"            # or postgres, typesense, tantivy, memory
url = "http://127.0.0.1:7700"     # meilisearch and typesense
prefix = "crm_"                   # put in front of every index name
queue = "search"                  # the queue sync jobs run on
sync = "queued"                   # queued | inline | off
chunk_size = 500                  # rows per batch for search:import

A queue other than the default one needs a worker: add it to queue.queues (queues = ["default", "search"]), or run navree queue:work --queue search next to your other workers.

Keep the engine's API key out of the TOML file. In .env:

APP_SEARCH__KEY=masterKey-from-your-meilisearch-server

Every key can come from the environment: prefix it with APP_ and use __ between levels, as in APP_SEARCH__ENGINE=typesense or APP_SEARCH__TANTIVY__PATH=/var/lib/crm/search. An invalid engine name, a missing url for a server engine or a missing key for Typesense fails the boot with a message that names the key.

To configure the module in code, pass an engine, or full SearchOptions, to for_root:

use navree::search::{Engine, SearchOptions};

// The engine alone, every other option at its default:
Navree::app().module(SearchModule::for_root(Engine::tantivy("storage/search")))

// Or the whole options struct:
Navree::app().module(SearchModule::for_root(SearchOptions {
    engine: Engine::meilisearch("http://127.0.0.1:7700").key("masterKey"),
    prefix: "crm_".to_owned(),
    ..SearchOptions::default()
}))

At boot the module pushes every registered index's settings to the engine. An engine that is down at boot doesn't stop the app: the failure is logged and the settings are pushed on first use.

Choosing an engine

[search]
engine = "postgres"

[search.postgres]
language = "english"   # the text search configuration
trigram = true         # typo tolerance with pg_trgm

[search]
engine = "meilisearch"
url = "https://search.internal:7700"
# key: APP_SEARCH__KEY

[search]
engine = "typesense"
url = "https://typesense-1.internal:8108"
nodes = ["https://typesense-2.internal:8108", "https://typesense-3.internal:8108"]
# key: APP_SEARCH__KEY (required)

[search]
engine = "tantivy"

[search.tantivy]
path = "storage/search"     # one directory per index
memory_budget_mb = 64       # writer buffer; less than 15 counts as 15

Making an entity searchable

Derive Searchable on a SeaORM model and mark the fields that go into the index. This is the CRM's contact, with search added:

use navree::search::Searchable;
use sea_orm::entity::prelude::*;
use serde::Serialize;

use crate::enums::ContactStatus;

#[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel, Serialize, Searchable)]
#[sea_orm(table_name = "contacts")]
#[searchable(name = "Contact", soft_delete = "deleted_at")]
pub struct Model {
    #[sea_orm(primary_key, auto_increment = false)]
    pub id: Uuid,
    #[searchable(boost = 3)]
    pub first_name: String,
    #[searchable(boost = 3)]
    pub last_name: String,
    #[searchable(boost = 2)]
    pub email: Option<String>,
    #[searchable]
    pub phone: Option<String>,
    #[searchable]
    pub title: Option<String>,
    #[searchable(filter)]
    pub status: ContactStatus,
    #[searchable(filter)]
    pub company_id: Option<Uuid>,
    #[searchable(filter)]
    pub owner_id: Uuid,
    #[searchable(sort)]
    pub created_at: DateTimeUtc,
    pub updated_at: DateTimeUtc,
    pub deleted_at: Option<DateTimeUtc>,
}

The derive implements Searchable for the module's Entity, so you search with the same type your access rules use (Contact). The primary key is always stored; fields without #[searchable] stay out of the index, and the entity needs a single-column primary key.

Full-text and embedded fields go into the document as serde serializes them. Filter and sort fields hold the column's database value, the form access rules compare: an enum's stored string ("closed_won", whatever its serde name), a UUID hyphenated, a timestamp in UTC. Filter with those values: filter("stage", "closed_won"). A field's column is the field's name, or its #[sea_orm(column_name = "..")].

Then register the entity with SearchModule::for_feature, in the module that owns it:

#[module(
    imports = [
        AccessModule::for_feature::<Contact>(),
        SearchModule::for_feature::<Contact>(),
        SearchModule::for_feature::<Company>(),
        SearchModule::for_feature::<Deal>(),
    ],
    register = Self::wire,
)]
pub struct CrmCoreModule;

Registration does three things: it adds the index to search:import and search:status, pushes the index settings to the engine at boot, and subscribes the entity's changes to the sync observer. Two entities can't share an index name or a name; the boot fails if they do.

Documents with related data

The derive builds a document from the row alone. When a document needs data from other tables, implement Searchable by hand. A deal is easier to find by its company's name:

use crm_core::entities::{company, deal};
use navree::BoxFuture;
use navree::search::{Document, IndexSettings, SearchContext, Searchable};
use sea_orm::EntityTrait;

impl Searchable for deal::Entity {
    fn settings() -> IndexSettings {
        IndexSettings::new("Deal", "deals")
            .text("title", 3)
            .text("company_name", 2)
            .filter("stage")
            .filter("owner_id")
            .filter("company_id")
            .sort("amount_cents")
            .sort("expected_close_date")
    }

    fn document<'a>(
        deal: &'a deal::Model,
        ctx: &'a SearchContext,
    ) -> BoxFuture<'a, Result<Option<Document>, navree::BoxError>> {
        Box::pin(async move {
            let company = company::Entity::find_by_id(deal.company_id)
                .one(ctx.conn())
                .await?;
            Ok(Some(
                Document::new(deal.id)
                    .field("title", &deal.title)
                    .field("company_name", company.map(|c| c.name))
                    .field("stage", deal.stage.as_str())
                    .field("owner_id", deal.owner_id)
                    .field("company_id", deal.company_id)
                    .field("amount_cents", deal.amount_cents)
                    .field("expected_close_date", deal.expected_close_date),
            ))
        })
    }
}

Return Ok(None) to keep a record out of the index (an archived deal, a draft). A record that returns None after it was indexed is removed. Only the fields settings() declares reach the index. A hand-written settings() names its soft-delete and workspace columns itself: .soft_delete("deleted_at"), .workspace("workspace_id"). Give a filter field the column's database value, as deal.stage.as_str() does here, so that access rules on that column filter in the engine exactly.

When a company is renamed, its deals' documents are stale. Queue them for a re-sync from an event listener, or wherever you rename the company:

let ids: Vec<Uuid> = deals_of_company.iter().map(|d| d.id).collect();
search.reindex::<Deal>(ids).await?;

Keeping indexes in sync

SearchModule registers a change observer on the unit of work. When a unit of work that wrote a searchable entity through Repo commits, the observer queues one search.sync job per entity type (and per workspace) with the changed ids. The job is part of the commit: with the database queue driver its row is written in the same transaction, and the memory driver queues it right after the commit. So:

• a request that fails (status 400 or above) rolls back its writes and queues nothing;
• the worker runs after the commit, so it always indexes committed data;
• a burst of writes to the same records collapses into one sync, since sync jobs are unique per entity and ids for search.unique_for_secs (5 seconds).

The job loads the current rows and upserts or deletes their documents. It retries with backoff like any job, so an engine that is down for a minute catches up on its own.

With queued and no QueueModule in the app, sync jobs go to the app's dyn Dispatcher if it has one; with neither, the module syncs inline and logs a warning once.

Writes that bypass Repo (raw SQL, update_many, a migration that backfills a column) are not observed. Re-sync those records with search.reindex::<E>(ids), or rebuild the index with search:import. reindex queues a sync job with sync = "queued" and syncs the records right away otherwise (off included).

To pause syncing for a block of work, such as a bulk import that ends with a full reindex, wrap it. Changes committed while the future runs are not synced, so commit inside it:

search
    .without_syncing(async {
        let uow = db.uow(None, None).await;
        import_contacts_from_csv(&uow, &file).await?;
        uow.commit().await
    })
    .await?;
search.import::<Contact>().await?;

Importing and rebuilding indexes

search:import indexes every row of an entity, in batches of search.chunk_size:

navree search:import Contact

Importing Contact into crm_contacts: 12,480 rows
[==========================================] 12480/12480  2.1s
Imported 12,480 documents (37 skipped by the document filter).

Run search:import --all --fresh after you change a model's searchable fields, and as part of a deploy that changes them. search.import::<E>() and search.import_fresh::<E>() do the same from code, for an admin screen or a scheduled task; both return an ImportReport with the rows read, documents written and rows skipped.

Searching

Inject Arc<Search> and start a query with the entity type and the text. paginate takes a PageQuery and returns a Page of models, in relevance order:

use crm_core::Contact;
use navree::prelude::*;
use navree::search::Search;
use serde::Deserialize;

#[derive(Debug, Deserialize, garde::Validate)]
pub struct SearchContacts {
    #[garde(length(min = 1, max = 200))]
    pub q: String,
    #[garde(skip)]
    pub status: Option<String>,
}

#[injectable]
pub struct ContactSearchController {
    search: Arc<Search>,
}

#[controller("/contacts/search", version = "1")]
impl ContactSearchController {
    /// `GET /api/v1/contacts/search?q=ada&status=customer`
    #[get("/")]
    #[can(Read, Contact)]
    async fn index(
        &self,
        scope: Accessible<Contact, Read>,
        Valid(Query(input)): Valid<Query<SearchContacts>>,
        page: PageQuery,
    ) -> Result<Page<ContactResource>> {
        let mut query = self.search.query::<Contact>(&input.q).accessible(scope);
        if let Some(status) = &input.status {
            query = query.filter("status", status);
        }
        let contacts = query.paginate(page).await?;
        Ok(contacts.map(|contact| ContactResource::new(contact, None, None)))
    }
}

{
  "data": [
    { "__type": "Contact", "id": "0199c1d2-...", "full_name": "Ada Lovelace", "email": "[email protected]" }
  ],
  "meta": { "current_page": 1, "per_page": 15, "total": 1, "last_page": 1 }
}

The engine returns ids and scores. Navree loads the rows from the database in one query and keeps the engine's order, so you always return current data, and your resource types work unchanged. Rows deleted since the last sync are dropped from the page.

Filters and sorting

Filters apply to #[searchable(filter)] fields; sorts to #[searchable(sort)] fields. Anything else fails with SearchError::NotFilterable or NotSortable before a request reaches the engine.

let deals = search
    .query::<Deal>("renewal")
    .filter("stage", "negotiation")
    .filter_in("owner_id", [ada_id, grace_id])
    .range("amount_cents", 1_000_000..)
    .sort_desc("amount_cents")
    .paginate(page)
    .await?;

Values compare by JSON type: a number never equals a string. Dates and timestamps are stored in the index as Unix milliseconds, so ranges and sorts on them work on every engine; pass dates (NaiveDate, DateTime<Utc>) or their ISO strings.

An empty query text matches every document, so a filtered empty query is a fast list: search.query::<Deal>("").filter("owner_id", me).sort_desc("expected_close_date"). ids(limit) returns the first matches' primary keys and count() the number of matches.

Access control

accessible(scope) takes the Accessible<E, Read> extractor from authorization. Navree turns the user's ability conditions into engine filters when they only use filterable fields on boolean, number, string, UUID or enum columns (the usual owner_id = me rule becomes filter("owner_id", me)), with the same rule order and the same comparisons as SQL lists, so pages stay full and totals stay exact. Other conditions are applied when the rows are loaded, with the same SQL filter as Accessible lists: conditions on fields the index doesn't hold, and on date and time columns, which the index keeps in milliseconds while the database compares finer. Records the user can't read are then dropped from the page, and its total can still count them. Mark the columns your rules use as filter to keep totals exact.

A rule the engine can't evaluate exactly takes the same path to the rows: on Meilisearch, a range on a text column (it compares numbers only); on Typesense, !=, not_in and null tests (it doesn't index missing values) and ranges on text; with Qdrant behind similar or hybrid, equality on a float. The PostgreSQL, Tantivy and memory engines run every translated rule themselves.

Hits and highlights

hits returns each record with its score and the matched text, marked up for the search box:

let page = search.query::<Contact>("lovlace").hits(page).await?;
for hit in &page.data {
    println!("{} {:?}", hit.score, hit.highlights.get("last_name"));
    // 0.94 Some("<mark>Lovelace</mark>")
}

Hit<T> has record, score (0 to 1, higher is better) and highlights, a map of field to its text, HTML-escaped, with <mark> around the matches; fields without a match are left out. Change the tags with .highlight_tags("<em>", "</em>"). Hit<T> serializes as the record's fields plus _score and _highlights, and hit.map(f) converts the record.

Searching several entities at once

The command palette searches contacts, companies and deals in one call. across runs one query per index in parallel and merges the results by score:

let results = search
    .across("ada")
    .include::<Contact>(contacts_scope)
    .include::<Company>(companies_scope)
    .include::<Deal>(deals_scope)
    .limit(10)
    .run()
    .await?;

for hit in results {
    // hit.kind: "Contact" | "Company" | "Deal"; hit.id; hit.title; hit.score
}

Each include takes that entity's Accessible scope. A result carries the name of its index as kind, the first full-text field as title and its highlights; call hit.load::<Contact>(&uow) for the whole record (None for a hit of another kind).

Workspaces

In an app with workspaces and row isolation, a searchable entity that implements WorkspaceScoped stores its workspace column (workspace_id by default) in every document. The derive finds the column; a hand-written settings() names it with .workspace("workspace_id").

Every query on such an index names the workspace, as WorkspaceRepo does for rows. In a request, pass the CurrentWorkspace id; in jobs and commands, the workspace you work in:

search.query::<Contact>("ada").workspace(ws.id()).paginate(page).await?;

A query without a workspace fails with SearchError::WorkspaceRequired, so one workspace's search never returns another's records. across and similar take .workspace(id) the same way. Sync jobs carry their records' workspace, and on PostgreSQL they read the rows inside it, so row-level security policies hold; search:import imports workspace by workspace there.

All workspaces share one index per entity, filtered by workspace. Schema and database isolation are not supported by search: the sync reads the app's own tables.

Vector search

Full-text search finds the words people type. Vector search finds records that mean the same thing: "customers unhappy with onboarding" finds the contact whose notes say "struggled to get the team set up". Mark the fields that describe a record with #[searchable(embed)]:

#[searchable(boost = 2, embed)]
pub title: String,
#[searchable(embed)]
pub notes: Option<String>,

On each sync, the embedded fields are joined into one text, one field per line, and turned into a vector by the app's embeddings provider: an Arc<dyn Embeddings> the app registers, usually a wrapper around an embedding model's API:

use navree::BoxFuture;
use navree::prelude::*;
use navree::search::Embeddings;

struct ModelEmbeddings { /* the model's client */ }

impl Embeddings for ModelEmbeddings {
    fn embed<'a>(&'a self, texts: &'a [String]) -> BoxFuture<'a, Result<Vec<Vec<f32>>, BoxError>> {
        Box::pin(async move {
            // One vector per text, `search.vectors.dimensions` numbers each.
            todo!()
        })
    }
}

// In a module's `register`:
m.value_arc(Arc::new(ModelEmbeddings { /* .. */ }) as Arc<dyn Embeddings>);

Vectors are stored in the configured vector store:

[search.vectors]
store = "pgvector"        # pgvector | qdrant | memory
dimensions = 1536         # must match the embedding model
metric = "cosine"         # cosine | dot | euclidean

# store = "qdrant"
# url = "http://127.0.0.1:6333"   # the REST port; APP_SEARCH__VECTORS__KEY for the API key

With vectors on, the boot fails without an Embeddings provider. With pgvector, the store enables the vector extension and creates a search_vectors table, with an embedding vector(1536) column and an HNSW index, the first time it is used. With Qdrant, each index gets a collection of the same name. The memory store is for tests.

Query by meaning with similar, or blend both kinds of relevance with hybrid:

// Nearest records to a text:
let similar = search
    .similar::<Contact>("customers unhappy with onboarding")
    .filter("status", "customer")
    .limit(10)
    .run()
    .await?;

// Nearest records to another record:
let like_this = search.similar_to::<Deal>(deal.id).limit(5).run().await?;

// Keyword and vector scores blended, 0.0 = keywords only, 1.0 = vectors only:
let deals = search.query::<Deal>("renewal risk").hybrid(0.4).paginate(page).await?;

run() returns Hits, nearest first, with the similarity (0 to 1) as score. Filters, workspaces and accessible work the same for vector queries. hybrid scores the best candidates of each side (at least 100, or as many as the page needs) as (1 - weight) * keyword + weight * vector. Embeddings are cached in memory by text, so re-syncing an unchanged record costs no call to the provider.

Testing

Use the memory engine with inline syncing. Writes through Repo are then searchable as soon as the unit of work commits, and nothing leaves the process:

use crm_testkit::{Role, TestCrm};
use navree::search::Search;
use serde_json::json;

#[tokio::test]
async fn finds_contacts_by_a_misspelled_name() {
    let crm = TestCrm::builder()
        .app(crm_api::app)
        .config("search.engine", "memory")
        .config("search.sync", "inline")
        .build()
        .await;
    let rep = crm.login_as(Role::Rep1).await;

    rep.post("/api/v1/contacts")
        .json(&json!({ "first_name": "Ada", "last_name": "Lovelace" }))
        .send()
        .await
        .assert_status(201);

    crm.get::<Search>().await.flush_pending().await.unwrap();

    rep.get("/api/v1/contacts/search")
        .query(&[("q", "lovlace")])
        .send()
        .await
        .assert_status(200)
        .assert_json_path("meta.total", json!(1))
        .assert_json_path("data.0.full_name", json!("Ada Lovelace"));
}

• flush_pending() waits until the inline syncs queued so far have finished. With sync = "queued", run the queue instead: crm.run_queue().await, or app.get::<Queue>().await.run_pending().await in a plain TestApp.
• The memory engine supports every query method, typo tolerance included, and ranks with the same field boosts, so tests and production agree on what matches. Exact scores differ between engines; assert on order and membership, not on _score.
• To check that a write queues a sync without running it, install a QueueFake and assert on the sync job, navree::search::SyncDocuments (job name search.sync): fake.assert_dispatched::<SyncDocuments>(|job| job.index == "contacts").
• search.assert_indexed::<Contact>(id) and search.assert_not_indexed::<Contact>(id) check the index directly.
• For vector search, set search.vectors.store = "memory" and register a fake Embeddings with override_value(Arc::new(fake) as Arc<dyn Embeddings>).
• To test a hand-written document(), build a SearchContext::new(tx, app.ctx().clone(), None) around a transaction.

Every engine implements the SearchEngine trait, which you can implement for an engine of your own and plug in with Engine::custom(Arc::new(engine)). Override its filters_exactly(filter) to return false for filter shapes your engine doesn't evaluate as the SQL filter does; access rules of those shapes are then applied to the loaded rows.

Reference

Configuration

The Search service

Errors

A search that the engine can't answer returns SearchError. With ? in a handler it becomes Navree's error shape: an invalid filter or sort field, or a filter the engine can't run, is a 400; an unreachable or timed-out engine is a 503:

{ "statusCode": 503, "message": "Search is unavailable. Try again shortly.", "error": "Service Unavailable", "code": "service_unavailable" }

The sync jobs log every failure at target navree::search with the index, the ids and the engine's answer, and retry.