paulserban.eu

Writing Edition

Paul Serban

Writing, snippets & book notes

Vol. 2026

August 08, 2026

Federated Data Ingestion Systems: Architecture for Multi-Source, Access-Controlled Retrieval

How to design connectors, normalization pipelines, hybrid storage, and policy-enforced serving layers for data that lives everywhere

Introduction

Most organizations don't have a data problem so much as a data geography problem. Customer records sit in Salesforce, product telemetry streams through Kafka, support tickets live in Zendesk, contracts sit in SharePoint, and internal wikis sprawl across Confluence and Notion. Each system has its own schema, its own authentication model, its own notion of "recently updated", and its own idea of who is allowed to see what. A federated data ingestion system is the architecture that reconciles all of this: it pulls data from heterogeneous sources, reshapes it into something queryable, and serves it back out without silently leaking permissions that existed in the source system but got lost along the way.

This matters more today than it did five years ago because retrieval-augmented generation (RAG) systems, internal search tools, and analytics platforms all depend on the same underlying need - unified, trustworthy access to distributed data. The difficulty isn't moving bytes from one place to another; plenty of ETL tools do that well. The difficulty is doing it in a way that preserves semantic meaning, supports both structured filtering and fuzzy semantic search, and never accidentally grants a user visibility into a document they couldn't open in the source system. This article walks through the four layers that make up a federated ingestion architecture - connectors, normalization/enrichment, hybrid storage, and access-controlled serving - with enough implementation detail to be useful on a real project.

The Problem: Why Federated Ingestion Matters

Centralizing data sounds simple in a planning meeting and becomes genuinely hard the moment you touch a second source system. Every connector you add brings its own rate limits, pagination quirks, webhook reliability guarantees (or lack thereof), and authentication lifecycle. A Jira connector that works flawlessly in a sandbox will eventually hit an API deprecation, a token expiry edge case, or a tenant with ten million issues that breaks your assumed pagination strategy. Multiply this by a dozen sources and the ingestion layer becomes one of the most operationally fragile parts of the stack, even though it's rarely where teams invest the most design effort up front.

The second-order problem is semantic drift. A "customer" in your CRM is not the same shape as a "customer" in your billing system, and a "status" field in one ticketing tool doesn't map cleanly onto another. Without a normalization layer, you end up with a storage layer that technically contains everything but is practically unqueryable, because nothing agrees on field names, units, timezones, or identifiers. This is the same problem data warehousing has wrestled with for decades, but it's compounded in modern systems by the addition of unstructured content - PDFs, transcripts, chat logs - that needs chunking, embedding, and enrichment rather than just column mapping.

The third problem, and the one most often underestimated, is access control. Source systems encode permissions in wildly different ways: Google Drive uses sharing lists, Confluence uses space permissions plus page restrictions, Salesforce uses a combination of profiles, roles, and sharing rules. When you ingest content from these systems into a central store, you are also implicitly responsible for ingesting - and continuously syncing - the permission model that governed it. Getting this wrong doesn't just produce bad search results; it produces data breaches where a user queries a central index and gets back a snippet of a document they were never authorized to see in the source.

Architecture Overview

A federated ingestion system is best understood as a pipeline with four distinct layers, each with a different failure mode and a different set of engineering concerns. The source connector layer is responsible for extraction: authenticating against external systems, handling pagination and rate limits, and emitting both the raw content and its associated permission metadata. The normalization and enrichment layer takes that raw, heterogeneous output and transforms it into a canonical schema, adding derived fields like embeddings, extracted entities, and classification tags along the way.

The storage layer persists both the structured and unstructured representations in a way that supports hybrid retrieval - meaning a single query can combine keyword/filter-based search with semantic vector search and return a unified, ranked result set. Finally, the retrieval and serving layer is the only component that end users or downstream applications ever touch directly, and it is responsible for enforcing access control on every single request, not just at ingestion time. Treating these as four separate concerns, each independently scalable and independently testable, is what keeps the system maintainable as the number of sources grows from three to thirty.

Source Connectors Layer

Connectors are the most numerous and most operationally demanding part of the system, because every external API has opinions you didn't ask for. A well-built connector framework separates three concerns: authentication and token refresh, the extraction strategy (full sync versus incremental/CDC), and the mapping of raw API responses into a common "document envelope" format before anything downstream has to deal with source-specific quirks. Change Data Capture tools like Debezium are the standard approach for databases, streaming row-level changes out of MySQL, PostgreSQL, or MongoDB via their native replication logs rather than polling. For SaaS APIs without CDC support, incremental sync usually means tracking a high-water mark (an updated-since timestamp or a cursor) per source and per tenant.

Open-source connector frameworks such as Airbyte and the Singer/Meltano ecosystem are worth evaluating before building your own, because they already encode a huge amount of hard-won knowledge about specific APIs' rate limits and pagination formats. That said, most teams end up writing custom connectors for a handful of business-critical or unusual sources, and it's worth designing your own connector interface so that custom and off-the-shelf connectors can coexist behind the same contract.

The critical design decision at this layer is what the connector emits besides content. Every connector should emit a normalized "access context" alongside the payload - the source system's ACL representation, translated into a common vocabulary (user IDs, group IDs, visibility scope) - even if nothing downstream uses it yet. Retrofitting permission metadata after a system is already in production is far more expensive than capturing it from day one, because by the time you notice the gap, you've already indexed content whose access boundaries you can no longer reconstruct without re-crawling the source.

// Common connector contract that all source-specific connectors implement
interface AccessContext {
  sourceSystem: string;              // e.g. "confluence", "salesforce"
  principals: {
    type: "user" | "group" | "public";
    id: string;
  }[];
  visibility: "private" | "restricted" | "internal" | "public";
}

interface RawDocument {
  sourceId: string;
  sourceSystem: string;
  content: string | Buffer;
  contentType: "text" | "html" | "pdf" | "structured";
  rawMetadata: Record<string, unknown>;
  accessContext: AccessContext;
  fetchedAt: string; // ISO 8601
}

interface Connector {
  name: string;
  authenticate(): Promise<void>;
  // Incremental sync using a persisted cursor per tenant/source
  fetchChanges(cursor: string | null): AsyncGenerator<RawDocument>;
  // Translates source-native ACLs into the common AccessContext shape
  resolveAccessContext(rawAcl: unknown): AccessContext;
}

// Example: Confluence connector sketch
class ConfluenceConnector implements Connector {
  name = "confluence";

  async authenticate() {
    // OAuth2 client-credentials or API token refresh handled here
  }

  async *fetchChanges(cursor: string | null) {
    let start = cursor ? parseInt(cursor, 10) : 0;
    const pageSize = 50;

    while (true) {
      const res = await this.fetchPage(start, pageSize); // calls Confluence REST API
      if (res.results.length === 0) break;

      for (const page of res.results) {
        yield {
          sourceId: page.id,
          sourceSystem: this.name,
          content: page.body.storage.value,
          contentType: "html" as const,
          rawMetadata: { title: page.title, spaceKey: page.space.key },
          accessContext: this.resolveAccessContext(page.restrictions),
          fetchedAt: new Date().toISOString(),
        };
      }
      start += pageSize;
    }
  }

  resolveAccessContext(rawAcl: any): AccessContext {
    // Map Confluence space + page-level restrictions into common principals
    const principals = (rawAcl?.restrictions?.user?.results ?? []).map((u: any) => ({
      type: "user" as const,
      id: u.accountId,
    }));
    return {
      sourceSystem: this.name,
      principals,
      visibility: principals.length > 0 ? "restricted" : "internal",
    };
  }

  private async fetchPage(start: number, limit: number) {
    // actual HTTP call omitted
    return { results: [] as any[] };
  }
}

Normalization & Enrichment Pipeline

Once raw documents leave the connector layer, they need to converge on a single canonical schema before anything downstream can reason about them. Normalization typically operates on two axes: structural normalization (field names, types, units, timezones) and semantic normalization (entity resolution, so "Acme Corp", "ACME Inc", and "acme-corp" in three different systems collapse into one canonical entity ID). Tools like dbt are widely used for the structural side when the targets are tabular, while entity resolution for unstructured or semi-structured records often relies on a combination of deterministic matching (shared external IDs, email domains) and probabilistic matching (fuzzy name/address matching) depending on how much precision the downstream use case demands.

Enrichment is where the pipeline adds value rather than just reshaping it. This is the stage where you chunk long documents for retrieval, generate vector embeddings, extract named entities, run classification models to tag sensitivity or document type, and attach provenance metadata. A critical, often-skipped practice here is data quality validation before enrichment runs - frameworks like Great Expectations let you assert schema and value constraints on incoming data so that a malformed upstream payload fails loudly in a staging buffer instead of silently corrupting embeddings or enrichment output downstream. Streaming frameworks such as Apache Flink or Spark Structured Streaming are commonly used to run this enrichment at scale, particularly when embeddings generation needs to be batched for throughput against an embedding model's API limits.

from dataclasses import dataclass
from typing import Optional
import hashlib

@dataclass
class NormalizedChunk:
    chunk_id: str
    document_id: str
    source_system: str
    text: str
    entity_tags: list[str]
    sensitivity: str  # "public" | "internal" | "confidential"
    embedding: Optional[list[float]] = None
    access_context_id: str = ""  # foreign key into the access context store


def chunk_document(text: str, max_tokens: int = 300, overlap: int = 50) -> list[str]:
    """Naive sliding-window chunker; production systems typically use
    sentence-aware or semantic chunking instead of fixed token windows."""
    words = text.split()
    step = max_tokens - overlap
    return [
        " ".join(words[i:i + max_tokens])
        for i in range(0, len(words), step)
        if words[i:i + max_tokens]
    ]


def normalize_and_enrich(raw_doc: dict, embed_fn, classify_fn) -> list[NormalizedChunk]:
    chunks = chunk_document(raw_doc["content"])
    results = []

    for i, chunk_text in enumerate(chunks):
        chunk_id = hashlib.sha256(
            f"{raw_doc['sourceId']}-{i}".encode()
        ).hexdigest()[:16]

        entities = classify_fn(chunk_text, mode="ner")
        sensitivity = classify_fn(chunk_text, mode="sensitivity")
        embedding = embed_fn(chunk_text)

        results.append(NormalizedChunk(
            chunk_id=chunk_id,
            document_id=raw_doc["sourceId"],
            source_system=raw_doc["sourceSystem"],
            text=chunk_text,
            entity_tags=entities,
            sensitivity=sensitivity,
            embedding=embedding,
            access_context_id=raw_doc["accessContext"]["sourceSystem"] + ":" + raw_doc["sourceId"],
        ))

    return results

Storage Layer for Hybrid Retrieval

"Hybrid retrieval" means a query can combine exact filtering (metadata, keywords, date ranges) with approximate semantic matching (vector similarity) and return results ranked by a combination of both signals. This has become the default expectation for any serious search or RAG system because pure keyword search misses paraphrased or conceptually related content, while pure vector search is weak at precise filtering - "show me contracts from Acme Corp signed after March" is a filtering problem, not a similarity problem. Systems like Elasticsearch and OpenSearch support this natively by combining BM25 lexical scoring with k-nearest-neighbor vector search in a single query, and dedicated vector databases such as Pinecone and Weaviate increasingly support metadata filtering alongside vector search for the same reason. PostgreSQL with the pgvector extension is another common choice when teams want hybrid retrieval without introducing a new database technology into their stack.

The storage layer for a federated system usually needs to serve two different access patterns well: point lookups and structured filtering for operational queries, and approximate nearest-neighbor search over embeddings for semantic retrieval. A common pattern is to store the canonical, enriched record in an analytical table format - Apache Iceberg or Delta Lake are the two dominant open table formats - while separately indexing the same content's embeddings and metadata in a search engine or vector store optimized for retrieval latency. The table format acts as the system of record and supports reprocessing (re-embedding with a new model, backfilling a new enrichment field) without re-crawling every source, while the retrieval index is treated as a derived, rebuildable artifact.

A frequently underestimated requirement at this layer is reconciling result ranking across the two retrieval modes. Reciprocal Rank Fusion (RRF) is a widely used, simple technique for merging ranked lists from lexical and vector search into one combined ranking without needing to normalize incomparable similarity scores directly. Whatever fusion method you choose, it needs to be tunable per use case - a legal document search tool and a customer support chatbot have very different tolerances for precision versus recall, and baking a single fixed fusion weight into the storage layer tends to produce retrieval quality that's mediocre for every consumer rather than good for any of them.

Retrieval/Serving Layer with Access Control

This is the layer where the architecture either earns trust or loses it. Access control enforced only at ingestion time - tagging a document with a sensitivity label once and never revisiting it - breaks the moment permissions change in the source system after ingestion, which happens constantly: an employee leaves, a document's sharing settings get tightened, a project folder gets restricted. The serving layer must evaluate access on every single request against the current state of the access model, not the state that was true when the content was indexed. This typically means the retrieval index stores a reference to an access policy rather than a frozen snapshot of permissions, and that reference is resolved against a live or near-live permission store at query time.

Attribute-Based Access Control (ABAC), as described in NIST SP 800-162, is a natural fit here because federated systems need to express conditions that pure role-based access control struggles with - "this user can see this document if they're in the same business unit and the document isn't tagged confidential and it's past its embargo date". Policy engines like Open Policy Agent (OPA) let you externalize these rules from application code and evaluate them consistently across every query path, including both the primary API and any batch export jobs. Where the underlying storage supports it, row-level security - available natively in PostgreSQL, for example - provides a second enforcement layer at the database boundary, so a bug in application-level filtering doesn't become a full data exposure.

// Query-time access enforcement: every retrieval call is filtered
// through a policy decision, never trusted from index metadata alone.

interface QueryContext {
  userId: string;
  groups: string[];
  tenantId: string;
}

interface RetrievedChunk {
  chunkId: string;
  text: string;
  score: number;
  accessContextId: string;
}

async function hybridSearch(
  query: string,
  ctx: QueryContext,
  opaClient: OpaClient,
  searchIndex: SearchIndex,
): Promise<RetrievedChunk[]> {
  // 1. Run hybrid (lexical + vector) retrieval, over-fetching candidates
  //    because some will be filtered out by access control downstream.
  const candidates = await searchIndex.hybridQuery(query, {
    tenantId: ctx.tenantId,
    limit: 100, // over-fetch; final result set will be smaller
  });

  // 2. Resolve access context IDs to their current permission state
  const accessContextIds = [...new Set(candidates.map(c => c.accessContextId))];
  const policyInputs = accessContextIds.map(id => ({
    userId: ctx.userId,
    groups: ctx.groups,
    resourceId: id,
  }));

  // 3. Evaluate policy decisions in a single batched call to OPA
  const decisions = await opaClient.batchEvaluate("data.retrieval.allow", policyInputs);
  const allowedIds = new Set(
    accessContextIds.filter((_, i) => decisions[i] === true)
  );

  // 4. Filter candidates and return only what the caller is authorized to see
  return candidates
    .filter(c => allowedIds.has(c.accessContextId))
    .slice(0, 20);
}

interface OpaClient {
  batchEvaluate(policyPath: string, inputs: unknown[]): Promise<boolean[]>;
}
interface SearchIndex {
  hybridQuery(query: string, opts: { tenantId: string; limit: number }): Promise<RetrievedChunk[]>;
}

Implementation Walkthrough: Putting the Layers Together

Consider a concrete scenario: an internal knowledge assistant that needs to answer questions by drawing on Confluence pages, Salesforce opportunity notes, and Zendesk tickets, while respecting each system's original permissions. The ingestion pipeline runs three connectors on independent schedules - Confluence polled every fifteen minutes using its "updated since" API parameter, Salesforce consuming Change Data Capture events over its Pub/Sub API, and Zendesk polled hourly via its incremental export endpoint. Each connector writes raw documents, including resolved access context, to a staging area (commonly a Kafka topic or an object storage landing zone) rather than writing directly to the final store, which gives you a replay point if a downstream enrichment step has a bug.

From staging, a stream processor picks up new and changed documents, runs them through the normalization and enrichment pipeline, and writes two outputs: the full enriched record to an Iceberg table (the system of record, queryable for audits and reprocessing) and the chunked, embedded representation to the hybrid search index. Crucially, the access context travels with the record at every stage - it's never summarized away into a single boolean "is this public" flag, because that discards information the policy engine needs later, such as which specific groups have access.

At query time, the serving layer doesn't ask "was this document public when we indexed it". It asks the policy engine, in real time, "can this specific user see this specific document right now", using the live access context reference stored alongside the chunk. This means a permission change in Salesforce - say, an opportunity getting restricted to a smaller sales team - takes effect on the very next query, not on the next full re-sync of the ingestion pipeline. The cost of this design is an extra network round-trip per query to the policy engine, which is why batching policy decisions (as shown in the code example above) and caching recent decisions for a short TTL are standard mitigations.

Trade-offs and Pitfalls

The most common failure mode in these systems is treating access control as a batch ingestion-time concern rather than a query-time concern, usually because it's simpler to implement and performs better under load. Teams that take this shortcut discover the problem only when an access review surfaces a document that's still retrievable weeks after its source permissions changed - by which point the fix requires auditing every index entry that might be stale, which is a far more expensive remediation than building query-time enforcement from the start. There is a genuine latency cost to real-time policy evaluation, and it's worth being honest about that cost rather than architecting it away by accident.

The second major trade-off is between freshness and system complexity. Fully real-time CDC-based ingestion from every source is attractive on paper but expensive to build and operate for long-tail sources that don't support CDC, have unreliable webhooks, or rate-limit aggressively. Most production systems end up with a tiered freshness model - near-real-time for high-value, high-change-rate sources, and scheduled batch sync for everything else - and this tiering should be an explicit architectural decision, documented and monitored, rather than something that happens implicitly because one connector was easier to build than another. Teams that don't make this trade-off explicit often end up with inconsistent user expectations about data freshness across different parts of the same product.

Best Practices

Schema evolution deserves deliberate handling from day one, because source systems change their APIs and your canonical schema will need to evolve alongside them. Using a schema registry (such as Confluate's Confluent Schema Registry, if your staging layer runs on Kafka with Avro or Protobuf) with enforced backward-compatibility rules prevents a source API change from silently breaking every downstream consumer of the normalized schema. Treat your canonical schema as a versioned, reviewed artifact, not an implicit agreement between the connector author and whoever happens to be maintaining enrichment at the time.

Observability across the pipeline is not optional once you have more than two or three sources, because failures tend to be partial and silent - a connector that's been failing for three days for one tenant, while every other tenant syncs fine, is exactly the kind of bug that doesn't trip a naive health check. Instrumenting every stage with structured tracing (OpenTelemetry is the standard here) and tracking per-source, per-tenant lag metrics lets you catch staleness before a user notices missing or outdated results.

Finally, test access control as rigorously as you test functional correctness. A policy engine with 100% unit test coverage on its rule logic can still fail in production if the data feeding it - the access context resolved by a connector - is wrong at the source. Building integration tests that simulate real permission changes end-to-end, from a source system mutation through to a query-time decision, catches the class of bug that unit tests on individual layers will never find, because the bug lives in the seams between layers rather than inside any single one.

Key Takeaways

  1. Capture access context in the connector layer from day one - retrofitting permissions after ingestion is far more expensive than designing for it up front.
  2. Keep the four layers (connectors, normalization/enrichment, storage, serving) independently deployable and testable; they fail in different ways and scale differently.
  3. Enforce access control at query time against live permission state, never solely from metadata frozen at ingestion time.
  4. Design for hybrid retrieval explicitly - lexical and vector search solve different problems, and fusion ranking needs to be tunable per use case.
  5. Make freshness trade-offs explicit and documented rather than letting them emerge accidentally from which connectors were easiest to build.

Conclusion

Federated data ingestion is, at its core, a problem of preserving meaning and trust as data crosses system boundaries it was never designed to cross. The technical challenges - connector reliability, schema normalization, hybrid indexing - are well-understood and solvable with existing tools. The harder, and more consequential, challenge is treating access control as a first-class, continuously-enforced property of the system rather than a label attached once and forgotten. Organizations that get this right end up with retrieval systems people actually trust enough to rely on; organizations that don't end up with impressive demos and a quiet, growing liability.

The architecture described here - connectors that preserve access context, a normalization and enrichment pipeline that adds real semantic value, hybrid storage that supports both precise filtering and semantic search, and a serving layer that re-evaluates permissions on every request - isn't the only way to solve this problem, but it reflects patterns that have proven durable across search, analytics, and RAG systems built on top of genuinely distributed data. The investment is front-loaded: it's slower to build than a simple "crawl everything into one index" approach. But it's the difference between a system that scales gracefully to thirty sources and one that needs to be rebuilt the first time a security review asks an uncomfortable question.

References