# How the Macro Search Service Integrates with OpenSearch for Full-Text Search

> Discover how the Macro search service integrates with OpenSearch for efficient full-text search. Learn about its indexing, bulk updates, and query execution capabilities.

- Repository: [Macro/macro](https://github.com/macro-inc/macro)
- Tags: how-to-guide
- Published: 2026-08-14

---

**The Macro platform delegates full-text search to an OpenSearch cluster through a reusable `OpensearchClient` wrapper that handles indexing, bulk updates, and query execution across all searchable entities.**

This deep dive examines how the `macro-inc/macro` repository implements end-to-end search functionality. The integration spans Rust-based backend services, TypeScript SDK clients, and infrastructure automation that provisions and maintains the search cluster.

## The OpensearchClient: Core Abstraction Layer

All OpenSearch operations flow through **`OpensearchClient`**, defined in [`crates/opensearch_client/src/lib.rs`](https://github.com/macro-inc/macro/blob/main/crates/opensearch_client/src/lib.rs). This crate abstracts the OpenSearch HTTP REST API into type-safe, instrumented Rust methods.

### Client Configuration

The client initializes from environment configuration and maintains an internal `reqwest::Client` for async HTTP calls:

- **Base URL**: Populated from `OPENSEARCH_URL` environment variable
- **HTTP client**: Reused across requests with connection pooling
- **Observability**: All methods wrapped in `tracing::instrument` spans
- **Error context**: Failures enriched with `rootcause` crate diagnostics

### Core Client Methods

| Method | Purpose | OpenSearch Endpoint |
|--------|---------|---------------------|
| `upsert_document` | Index single document | `PUT /{index}/_doc/{id}` |
| `bulk_upsert` | Batch index/update | `POST /_bulk` |
| `delete_document` | Remove from index | `DELETE /{index}/_doc/{id}` |
| `search` | Execute full-text query | `POST /{index}/_search` |

The `search` method accepts an index name and a raw OpenSearch DSL JSON body, returning deserialized response data for further processing.

## Data Ingestion: Keeping Indices Synchronized

Domain services push searchable content to OpenSearch through the client during their processing pipelines. Each searchable entity type has dedicated indexing logic.

### Document Indexing

The document processing pipeline in [`services/search_processing_service/src/process/document/raw_document.rs`](https://github.com/macro-inc/macro/blob/main/services/search_processing_service/src/process/document/raw_document.rs) demonstrates the typical upsert flow:

```rust
use opensearch_client::{OpensearchClient, upsert::documents::IndexedDocument};

#[tracing::instrument(skip(opensearch_client, db, s3_client))]
pub async fn upsert_document(
    opensearch_client: &OpensearchClient,
    search_extractor_message: &SearchExtractorMessage,
    db: &Database,
    s3_client: &S3Client,
) -> Result<()> {
    // Build typed payload matching the documents index mapping
    let indexed = IndexedDocument::from_message(search_extractor_message, db, s3_client).await?;
    
    // Bulk upsert handles both new documents and updates
    opensearch_client
        .bulk_upsert(&[indexed])
        .await
        .context("failed to bulk upsert document in OpenSearch")?;
    Ok(())
}

```

The `IndexedDocument` struct is defined in [`crates/opensearch_client/src/upsert/documents.rs`](https://github.com/macro-inc/macro/blob/main/crates/opensearch_client/src/upsert/documents.rs) and serializes to match the OpenSearch mapping exactly.

### Other Entity Types

The same pattern extends across all searchable data:

- **Projects**: [`services/search_processing_service/src/process/project.rs`](https://github.com/macro-inc/macro/blob/main/services/search_processing_service/src/process/project.rs) builds `IndexedProject` payloads
- **Chat messages**: [`services/search_processing_service/src/process/chat.rs`](https://github.com/macro-inc/macro/blob/main/services/search_processing_service/src/process/chat.rs) handles conversation content
- **Emails**: [`services/search_processing_service/src/process/email/upsert.rs`](https://github.com/macro-inc/macro/blob/main/services/search_processing_service/src/process/email/upsert.rs) indexes message bodies and metadata
- **Users**: [`services/search_processing_service/src/process/user.rs`](https://github.com/macro-inc/macro/blob/main/services/search_processing_service/src/process/user.rs) maintains people directory

Each module constructs domain-specific typed payloads that align with index mappings defined in the infrastructure automation.

## Query Execution: From API to Results

Search requests traverse multiple layers before reaching OpenSearch and returning to users.

### Request Flow Architecture

```

Web UI / SDK ──► Search Service Client ──► Search Processing Service ──► OpensearchClient.search()
                                     │
                                     ▼
                              OpenSearch Cluster
                                     │
                                     ▼
                              SearchResponse (JSON)

```

### The Search Service Handler

The backend endpoint in [`services/search_processing_service/src/process/mod.rs`](https://github.com/macro-inc/macro/blob/main/services/search_processing_service/src/process/mod.rs) receives requests and orchestrates query execution:

```rust
#[axum::debug_handler]
async fn handle_search(
    State(state): State<AppState>,
    Json(req): Json<SearchRequest>,
) -> Result<Json<SearchResponse>, ServiceError> {
    // Build OpenSearch DSL from user request parameters
    let dsl = build_search_dsl(&req);
    
    // Execute against configured index
    let raw = state.opensearch_client
        .search(&state.search_index, &dsl)
        .await?;
    
    // Transform raw OpenSearch response to API format
    let resp = SearchResponse::from_raw(raw);
    Ok(Json(resp))
}

```

### DSL Construction

The `build_search_dsl` function (internal to the search service) assembles:

- **`multi_match` queries** for full-text search across title, content, and description fields
- **`bool` filters** for entity type restrictions and access control
- **Aggregations** for facet counts (tags, owners, dates)
- **Pagination** parameters (`from`, `size`)
- **Highlighting** configuration for result snippets

### TypeScript SDK Interface

Client applications consume search through the typed SDK defined in [`packages/sdk/src/entities/search.ts`](https://github.com/macro-inc/macro/blob/main/packages/sdk/src/entities/search.ts):

```typescript
import { SearchService } from "@macro/sdk";

async function searchDocuments(query: string) {
  const response = await SearchService.search({
    query,
    entityTypes: ["Document"],          // Restrict to documents
    page: 1,
    pageSize: 20,
    facets: { tags: true, owners: true } // Request facet aggregations
  });

  console.log("Results:", response.hits);      // Matching documents
  console.log("Facets:", response.facets);     // Aggregation buckets
}

```

The front-end query wrapper in [`apps/web/src/lib/queries/soup/search.ts`](https://github.com/macro-inc/macro/blob/main/apps/web/src/lib/queries/soup/search.ts) provides React Query integration for caching and background refetching.

## Index Management and Local Development

### Infrastructure as Code

OpenSearch indices and mappings are defined in [`infra/stacks/opensearch/helpers/create_indices.sh`](https://github.com/macro-inc/macro/blob/main/infra/stacks/opensearch/helpers/create_indices.sh). This script:

- Creates indices for `documents`, `projects`, `messages`, `emails`, `users`
- Defines field mappings with appropriate analyzers
- Configures `text` fields for full-text search with standard analyzer
- Sets `keyword` fields for exact matching and faceting

### Local Provisioning

The `xtask` automation in [`tooling/xtask/crates/xtask_local/src/local/opensearch.rs`](https://github.com/macro-inc/macro/blob/main/tooling/xtask/crates/xtask_local/src/local/opensearch.rs) bootstraps development environments:

```rust
// Spans Docker-based OpenSearch container
// Executes create_indices.sh against local cluster
// Waits for health check before completing

```

This ensures developers have matching search infrastructure without manual setup.

## Performance and Reliability Patterns

**Bulk operations**: The `bulk_upsert` method batches changes to minimize HTTP overhead and OpenSearch refresh cycles.

**Instrumentation**: Every client method emits distributed traces with query timing, index names, and result counts.

**Error isolation**: Domain services catch OpenSearch failures, log structured errors, and continue processing—preventing indexing issues from disrupting core workflows.

**Mapping versions**: Typed payloads (`IndexedDocument`, `IndexedProject`) enforce compile-time alignment with index schemas, catching mapping mismatches during build rather than runtime.

## Summary

- **`OpensearchClient`** in [`crates/opensearch_client/src/lib.rs`](https://github.com/macro-inc/macro/blob/main/crates/opensearch_client/src/lib.rs) centralizes all OpenSearch HTTP communication
- **Domain services** in `services/search_processing_service/src/process/` push entity changes via `bulk_upsert` with typed payloads
- **Search queries** flow from SDK through service handlers that build OpenSearch DSL and return enriched responses
- **Index definitions** live in infrastructure automation with local development provisioning via `xtask`
- **Full-text capabilities** include multi-field matching, faceted filtering, result highlighting, and pagination

## Frequently Asked Questions

### How does Macro handle OpenSearch connection failures?

The `OpensearchClient` surfaces errors through the `rootcause` crate with structured context including the operation type, index name, and HTTP status. Domain services wrap calls in `anyhow::Context` to add business-level context (e.g., "failed to index document XYZ") while allowing processing pipelines to continue. Production deployments typically run OpenSearch with multiple nodes and use the client's retry-capable `reqwest` configuration for transient failures.

### Can search queries filter by multiple entity types simultaneously?

Yes. The `SearchRequest` struct accepts `entityTypes: string[]` and the DSL builder constructs a `bool` query with `should` clauses across indices. The search service may execute a single `_search` against multiple indices using comma-separated index names (e.g., `documents,projects,messages`) or parallel requests depending on the query complexity and facet requirements.

### What analyzers does Macro use for full-text search?

The [`create_indices.sh`](https://github.com/macro-inc/macro/blob/main/create_indices.sh) infrastructure script configures OpenSearch to use the **standard analyzer** for `text` fields, which applies word boundary tokenization and lowercase filtering. Keyword fields use the `keyword` analyzer for exact matching. Custom analyzers can be added to the provisioning script for specialized content like email addresses or code snippets.