Solve Stale Data and Access Rights in RAG Knowledge Bases

Knowledge Base Indexing for RAG: Solving Stale Data and Access Rights

Corporate knowledge bases are the primary context source for enterprise RAG systems. Retrieval-Augmented Generation (RAG) without an up-to-date index loses its value: in practice, we've encountered situations where RAG returned answers based on outdated content because Confluence pages hadn't been reindexed for weeks. Or worse — a user received data from a document they were not authorized to access. Both problems are solved with a proper indexing architecture for Confluence, Notion, and SharePoint knowledge bases.

With over 10 years of experience in enterprise search and proven methodologies, we have developed an incremental pipeline that processes only modified pages and strictly enforces access rights. Over more than 50 projects, we have built standard connectors and markup processing rules. Incremental synchronization is 6 times faster than full reindexing — reducing infrastructure costs and speeding up RAG answer updates. Get a consultation: we will show a demo on your data.

Why is incremental synchronization essential for RAG?

Full indexing every hour is expensive and slow. We use a "watermark approach": we store the timestamp of the last successful sync for each space/database. On the next run, we load only pages with last_modified > watermark. For Confluence, via the Atlassian REST API with the parameter expand=version,body.storage; for Notion, via a filter on last_edited_time.

Example connector for Confluence:

from atlassian import Confluence from datetime import datetime class ConfluenceIndexer: def __init__(self, url: str, username: str, api_token: str): self.confluence = Confluence( url=url, username=username, password=api_token, cloud=True # True for Atlassian Cloud ) self.watermark_store = WatermarkStore() def get_updated_pages(self, space_key: str) -> list[dict]: """Incremental load: only updated pages""" last_indexed = self.watermark_store.get(f"confluence:{space_key}") pages = self.confluence.get_all_pages_from_space( space=space_key, start=0, limit=100, expand='body.storage,metadata,version,ancestors' ) if last_indexed: pages = [ p for p in pages if datetime.fromisoformat(p['version']['when']) > last_indexed ] return pages def parse_page(self, page: dict) -> dict: from bs4 import BeautifulSoup from markdownify import markdownify # Confluence stores content in storage format (XHTML) html_content = page['body']['storage']['value'] soup = BeautifulSoup(html_content, 'html.parser') # Handle Confluence-specific tags for macro in soup.find_all('ac:structured-macro'): macro_name = macro.get('ac:name', '') if macro_name == 'code': # Code blocks -> markdown code blocks body = macro.find('ac:plain-text-body') lang = macro.find('ac:parameter', {'ac:name': 'language'}) code = body.get_text() if body else '' lang_str = lang.get_text() if lang else '' macro.replace_with(f'\n```{lang_str}\n{code}\n```\n') else: macro.decompose() text = markdownify(str(soup), heading_style="ATX") return { 'id': page['id'], 'title': page['title'], 'text': text, 'url': f"{self.confluence.url}/wiki{page['_links']['webui']}", 'space': page['space']['key'], 'ancestors': [a['title'] for a in page.get('ancestors', [])], 'labels': [l['name'] for l in page.get('metadata', {}).get('labels', {}).get('results', [])], 'last_modified': page['version']['when'], 'author': page['version']['by']['displayName'], # Access rights for permission-aware search 'restrictions': self._get_page_restrictions(page['id']) } 

A similar connector for Notion uses filtering by last_edited_time and recursive block extraction. Details can be found in the Notion API documentation.

Setting Up Permission-Aware Search

For integration with corporate IDPs (Azure AD, Okta), we proxy roles into the vector DB. Example implementation:

class PermissionAwareRetriever: def search(self, query: str, user_id: str, top_k: int = 5) -> list: # Get allowed document IDs for the user allowed_docs = self.permission_store.get_allowed_docs(user_id) # Vector search with permission filtering results = self.vector_store.similarity_search( query=query, filter={"doc_id": {"$in": allowed_docs}}, k=top_k ) return results 

Incremental synchronization every 15–60 minutes keeps the RAG system up-to-date without full reindexing of gigabytes of content. We use a "watermark approach" that reduces the processed data volume to 10–20% of a full dump. This results in cloud resource savings up to 40%, translating to over $5,000 per month for typical enterprise deployments by reducing token usage in incremental processing.

What's Included in the Indexing Work Scope?

• Connector documentation — description of each connector, its configuration, and processing logic.
• Permission mapping — table mapping IDP roles to vector DB groups.
• Chunking strategy — chunk size selection (token-based or semantic) with justification.
• MLOps pipeline — automatic synchronization with monitoring via Weights & Biases.
• Team training — two-hour workshop on index operations.

Chunking Strategy Selection

Embedding Model Recommendations

Typical Indexing Errors

• Missing Confluence macros — macros info, warning appear as blocks but their content is often lost. Our parser preserves them as blockquotes.
• Ignoring attachments — PDFs, DOCX files in Confluence/SharePoint contain crucial context. We attach an OCR pipeline.
• Lack of deduplication — identical pages in different spaces lead to embedding duplicates. A hash filter solves the problem.