Connectors & Content Ingestion

Overview

Connectors are how external content enters Qadra. A connector mirrors documents from a source system (Confluence, Google Drive, SharePoint, Slack, S3, or the gateway), feeds them through a processing pipeline, and loads the result into the vector index and SPO knowledge graph.

The design follows a Moveworks-inspired three-stage pipeline:

MIRROR (Connectors)  -->  PROCESS (Chunk + Enrich + Embed)  -->  LOAD (VectorIndex)

Everything is trait-based and swappable. Each connector implements the ContentConnector trait; the platform composes connectors, a DocumentProcessor, an EmbeddingProvider, and a VectorIndex behind abstract interfaces, so the ingestion core knows nothing about any specific source system.

The human-gated approval flow for documents (propose -> review -> approve -> ingest) and the ingestion internals (content analyzer, triple extraction, chunking, provenance) are covered in the Knowledge Graph chapter. This page covers the connectors, sync, and entitlements that sit in front of that pipeline.

The MIRROR -> PROCESS -> LOAD Pipeline

The MIRROR stage produces a SyncDelta (added / modified / deleted document IDs). Only changed documents flow downstream, so an incremental sync never re-processes the entire corpus.

The ContentConnector Model

A connector is a streaming source of documents. The ContentConnector trait exposes two core operations:

#![allow(unused)]
fn main() {
#[async_trait]
pub trait ContentConnector: Send + Sync {
    /// List documents, optionally only those changed since a timestamp (delta sync).
    async fn list_documents(&self, since: Option<DateTime<Utc>>) -> Result<Vec<DocumentState>>;

    /// Fetch full document content (streaming, zero-copy where possible).
    async fn fetch_documents(&self, ids: &[String]) -> Result<DocumentStream>;
}
}

Connectors are registered in the ServiceContainer via a ConnectorRegistry (HashMap<String, Arc<dyn ContentConnector>>), keyed by connector type. The supporting DocumentProcessor (chunk / enrich / embed) and VectorIndex (semantic search) traits complete the pipeline:

Available Connectors

The gateway connector is always available (it backs user uploads). The rest are opt-in per tenant and gated by entitlements (see below).

Sync Strategy

Each connector can be driven by three sync modes:

• Incremental sync is the common path: ask the source for everything changed since the last successful sync, build a SyncDelta, and process only the added/modified documents.
• Full sync runs daily to catch deletions and repair drift. Because it lists the entire corpus, it can detect documents that disappeared from the source (which incremental sync cannot reliably see) and rebuild the sync_state table from scratch.
• Webhook sync handles real-time updates: a single document is processed immediately on notification, without waiting for the next incremental window.

Sync history is recorded in the sync_runs table; per-document state lives in sync_state (primary key tenant_id + connector + source_id).

Change Detection

To decide whether a document actually changed, connectors compare source metadata in a strict priority order --- cheapest and most reliable first:

1. ETag comparison --- fastest, HTTP-native. If the source returns an ETag and it matches the stored one, the document is unchanged. No content fetch needed.
2. Version comparison --- for systems with explicit version numbers (Confluence, SharePoint). A higher version means the document changed.
3. Timestamp comparison --- the fallback. Compare last_modified against the stored synced_at.

Each document's tracked state is captured in DocumentState:

#![allow(unused)]
fn main() {
pub struct DocumentState {
    pub source_id: String,
    pub content_hash: String,        // BLAKE3 of content
    pub last_modified: DateTime<Utc>,
    pub etag: Option<String>,
    pub version: Option<u64>,
    pub synced_at: DateTime<Utc>,
}
}

A BLAKE3 content_hash is the final tiebreaker: even if ETag/version/timestamp suggest a change, an identical hash means the processed content is unchanged and downstream re-embedding can be skipped.

Zero-Copy Processing

Ingestion is engineered to avoid unnecessary allocations as content flows from source to vector store:

• DocumentContent is an enum of Mmap, Bytes, or Borrowed --- documents can be memory-mapped or borrowed with zero allocation on load.
• Chunk borrows from the document content (&'doc str) --- no string copies during chunking.
• EnrichedChunk uses Cow<'doc, str> --- it only allocates if a chunk is actually mutated during enrichment.

Supporting crates: memmap2 (memory-mapped files), bytes (zero-copy byte buffers), bumpalo (arena allocation), blake3 (fast content hashing).

Per-Tenant Connector Configuration

Connector configuration is per-tenant, stored in the tenant_connectors table and managed through the TenantConnectorStore trait:

Each configuration holds:

• Credentials --- one of OAuth2, ApiKey, Basic, or Aws, encrypted at rest.
• Sync config --- cron schedule, incremental interval, webhook settings, rate limits.
• Source filter --- connector-specific scoping: Confluence spaces, Google Drive folders, Slack channels, S3 prefixes.

Because credentials and source filters live per-tenant, two tenants can connect to the same source system (e.g. two Confluence instances) with completely isolated configuration and sync state.

Tenant Entitlements

Connectors and LLM providers are gated by service plans. Every tenant has an entitlement record defining what it can use and how much.

Service Plans

Three tiers --- free, pro, enterprise --- each defines:

• Limits: max_connectors, max_documents, api_requests, llm_tokens, storage, agents, pipelines, concurrent_workloads.
• Available services: which connectors and LLM providers are enabled.

Enforcement

Before any connector or service operation, the entitlement is checked:

#![allow(unused)]
fn main() {
// Before adding a connector / running a service operation:
let ent = entitlements.get(tenant_id).await?;
ent.can_add_connector(connector_type)?;  // checks service availability + limits
}

can_add_connector validates both that the connector type is available on the tenant's plan and that the tenant is under its max_connectors limit.

Override Hierarchy

Entitlements resolve in priority order, most specific first:

explicit service_overrides  >  limit_overrides  >  plan defaults

A tenant on the pro plan can be granted a single extra connector via a limit_override, or have a specific connector enabled via a service_override, without changing its plan.

Entitlement Schema

The TenantEntitlementStore trait exposes get, set_plan, set_limit_override, enable_service, and disable_service. Admin (Qadra employee) endpoints manage entitlements via GET/PUT/PATCH /admin/tenants/{id}/entitlements|plan|limits|services|usage|audit.

Feature-Flag Compilation

Beyond runtime entitlement checks, connectors are also gated at compile time via Cargo feature flags. A connector that is not compiled in cannot be enabled regardless of plan.

[features]
default     = ["postgres", "openrouter", "redis", "mongo", "pgvector", "connector-gateway"]
minimal     = ["sqlite", "ollama", "embed-local", "connector-gateway"]   # zero cloud deps
enterprise  = ["postgres", "openrouter", "pgvector", "redis", "mongo",
               "embed-openai", "connector-confluence", "connector-sharepoint",
               "connector-s3", "connector-gateway"]
full        = ["..."]  # everything

Two gates work together:

1. Build-time --- is_service_compiled(service) checks the active feature flags. The minimal build ships with only the gateway connector and zero cloud dependencies.
2. Runtime --- entitlements.get(tenant_id) checks the tenant's plan and overrides.

A connector must pass both gates to run: it must be compiled into the binary and enabled on the tenant's plan.