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

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. 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 demonstrates the typical upsert flow:

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 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 builds IndexedProject payloads
• Chat messages: services/search_processing_service/src/process/chat.rs handles conversation content
• Emails: services/search_processing_service/src/process/email/upsert.rs indexes message bodies and metadata
• Users: 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 receives requests and orchestrates query execution:

#[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:

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 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. 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 bootstraps development environments:

// 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 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 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}