[RFC] Database Index Schema Optimizations & Standardized Asynchronous Task Scheduling

[Shevilll]

RFC: Database Index Schema Optimizations & Standardized Asynchronous Task Scheduling

Status: Draft

Author: Antigravity (AI Contributor)
Target Repository: AOSSIE-Org/Devr.AI
Date: June 26, 2026

---

1. Executive Summary

As an AI-powered Developer Relations Assistant, Devr.AI is built to process high-throughput event streams from platforms like Discord, Slack, and GitHub in real-time. To maintain high responsiveness, sub-second latency, and horizontal scalability under high workloads, the backend infrastructure must be optimized both at the data persistence layer and the asynchronous task scheduling layer.

This RFC proposes two major system enhancements:

1. Database Index Schema Optimizations: Proactive B-Tree, GIN, and Partial indexes for PostgreSQL (Supabase) targeting high-read, high-write tables like interactions and repositories to eliminate Sequential Scans (Seq Scan).
2. Standardized Asynchronous Task Scheduling: Migrating ad-hoc cron scripts to a unified, database-backed, clustered task scheduling framework utilizing APScheduler with a PostgreSQL Job Store integrated into our existing FastAPI/Lifespan event loop.

---

2. Part 1: Database Index Schema Optimizations

2.1 The Problem

An audit of the current database schema file (create_db.sql) reveals several index gaps:

• The interactions Table: This is the most active table in the system, storing every message, comment, and PR event. However, no indexes are defined on user_id, repository_id, platform, created_at, or topics_discussed. Any query listing user interactions, filtering by platform, searching topics, or sorting by time forces a costly Seq Scan (O(N) search complexity).
• The repositories Table: Frequently searched by full_name (owner/repo) or owner/name to look up indexing state or retrieve metadata. Lack of index coverage on these columns degrades performance of LLM tool retrievals.
• The organization_integrations Table: Frequently queried by the Discord/Slack bots to verify which active organizations to serve. While it has single-column indexes, it lacks optimal composite index coverage for active platforms.

2.2 Proposed SQL Indexing Schema

We propose executing a database migration containing the following index declarations:

-- ====================================================================
-- DATABASE INDEX OPTIMIZATIONS FOR DEVR.AI
-- ====================================================================

-- --------------------------------------------------------------------
-- 1. Optimizations for 'interactions' table (High-Write/High-Read)
-- --------------------------------------------------------------------

-- Optimize foreign key joins and user history queries
CREATE INDEX IF NOT EXISTS idx_interactions_user_id 
    ON interactions(user_id);

CREATE INDEX IF NOT EXISTS idx_interactions_repository_id 
    ON interactions(repository_id) 
    WHERE repository_id IS NOT NULL;

-- Optimize temporal/activity-stream sorting (recent activities first)
CREATE INDEX IF NOT EXISTS idx_interactions_created_at_desc 
    ON interactions(created_at DESC);

-- Optimize lookup queries combining platform and platform-specific IDs (e.g. webhook deduplication)
CREATE INDEX IF NOT EXISTS idx_interactions_platform_lookup 
    ON interactions(platform, platform_specific_id);

-- Optimize thread and channel context retrieval (conversational memory queries)
CREATE INDEX IF NOT EXISTS idx_interactions_channel_thread 
    ON interactions(channel_id, thread_id) 
    WHERE channel_id IS NOT NULL;

-- Optimize classification and intent analytics
CREATE INDEX IF NOT EXISTS idx_interactions_classification 
    ON interactions(interaction_type, intent_classification) 
    WHERE interaction_type IS NOT NULL OR intent_classification IS NOT NULL;

-- Optimize search across topics_discussed array using GIN index (Generalized Inverted Index)
CREATE INDEX IF NOT EXISTS idx_interactions_topics_discussed_gin 
    ON interactions USING GIN (topics_discussed);


-- --------------------------------------------------------------------
-- 2. Optimizations for 'repositories' table (Metadata Lookups)
-- --------------------------------------------------------------------

-- Optimize fast case-insensitive full-name lookups from tools and agents
CREATE INDEX IF NOT EXISTS idx_repositories_full_name_lower 
    ON repositories(LOWER(full_name));

-- Optimize composite lookups by owner/name separately
CREATE INDEX IF NOT EXISTS idx_repositories_owner_name 
    ON repositories(owner, name);

-- Optimize partial index for indexing queues (fetching pending/failed repos)
CREATE INDEX IF NOT EXISTS idx_repositories_indexing_state 
    ON repositories(is_indexed, indexing_status) 
    WHERE is_indexed = false;


-- --------------------------------------------------------------------
-- 3. Optimizations for 'organization_integrations' table
-- --------------------------------------------------------------------

-- Optimize high-frequency active platform status checks by integrations
CREATE INDEX IF NOT EXISTS idx_org_integrations_platform_active 
    ON organization_integrations(platform, is_active) 
    WHERE is_active = true;

2.3 Query Performance Breakdown (Before & After)

Query 1: Fetching the latest 20 interactions for a user within a specific repository

SELECT * FROM interactions 
WHERE user_id = 'c1234567-89ab-cdef-0123-456789abcdef' 
  AND repository_id = '98765432-10fe-dcba-ba09-876543210fed'
ORDER BY created_at DESC 
LIMIT 20;

• Without Optimization: PostgreSQL performs a full Seq Scan on interactions, filtering out rows that do not match user_id and repository_id, then performs an in-memory or on-disk sort (Quick Sort) on created_at. Cost: O(N) where N is total interactions.
• With Optimization: PostgreSQL performs an Index Scan on idx_interactions_user_id or idx_interactions_repository_id, intersecting the two bitmap indexes, and uses the pre-sorted temporal index idx_interactions_created_at_desc to retrieve the rows immediately. Cost: O(log N) + constant lookup.

Query 2: Searching interactions discussing a specific array of topics (e.g. 'auth', 'database')

SELECT * FROM interactions 
WHERE topics_discussed @> ARRAY['auth', 'database']::TEXT[];

• Without Optimization: Forces a Sequential Scan and evaluates the array containment operator (@>) on every single row. Highly CPU intensive.
• With Optimization: Performs a Bitmap Index Scan on the GIN index idx_interactions_topics_discussed_gin, instantly pinpointing the matching rows without touching any unrelated disk pages. Performance gain is 100x to 1000x as data scales.

---

3. Part 2: Standardizing Asynchronous Task Scheduling

3.1 The Problem

Devr.AI needs to run scheduled background jobs:

1. Periodic synchronization of repository metrics (stars, forks, open issues).
2. Weekly compiler/cleanup of conversation summaries (conversation_context aggregation).
3. Automated health-checks of linked Discord and Slack integrations.
4. Token expiration monitoring for OAuth tokens.

Currently, Devr.AI uses aio-pika (RabbitMQ) for real-time reactive event handling, but lacks a centralized, reliable background scheduler. Writing infinite while True: await asyncio.sleep(86400) loops inside the web server is an anti-pattern:

• If the FastAPI process restarts, the sleep timer is reset.
• If multiple worker instances are scaled horizontally, each worker will run the same background task simultaneously, causing race conditions and database locks.
• There is no visibility, error logging, or manual trigger capabilities for scheduled jobs.

3.2 Proposed Architecture: APScheduler with PostgreSQL (Supabase) Job Store

We propose standardizing on APScheduler 3.x/4.x with a SQLAlchemy/PostgreSQL Job Store. This integrates perfectly with Devr.AI's Supabase-backed database layer.

flowchart TD
    subgraph FastAPI Web Process
        A[FastAPI App Lifespan] -->|Initializes| B(AsyncScheduler)
        B -->|Configures| C[(PostgreSQL Job Store)]
    end
    
    subgraph PostgreSQL Database
        C --- D[Table: apscheduler_jobs]
    end

    subgraph Cluster Deployment
        E[FastAPI Worker Instance 1] -->|Row Lock| D
        F[FastAPI Worker Instance 2] -->|Row Lock| D
    end

    B -->|Triggers Job| G[Execute Sync Job]
    G -->|Optionally Publishes Event| H[RabbitMQ Event Bus]

Loading

Key Benefits:

1. Persistence: Scheduled tasks are stored in a database table. If the servers crash, the scheduler picks up exactly where it left off, catching up on missed triggers (coalesce=True).
2. Distributed Locking (Cluster-Safe): If we scale to 3 FastAPI worker containers, APScheduler uses PostgreSQL's row-locking mechanisms to ensure that only one instance executes a scheduled job at a given time.
3. Async Native: Fully supports async/await syntax, allowing jobs to query databases or invoke agents using non-blocking I/O.

3.3 Database Table Definition for Jobs

APScheduler automatically generates its required table, but for completeness and integration with migrations, the underlying table structure in our PostgreSQL/Supabase DB is:

-- DDL for APScheduler persistence (if manually managed or audited)
CREATE TABLE IF NOT EXISTS apscheduler_jobs (
    id VARCHAR(191) PRIMARY KEY NOT NULL,
    next_run_time DOUBLE PRECISION,
    job_state BYTEA NOT NULL
);

COMMENT ON TABLE apscheduler_jobs IS 'Stores persistent scheduled tasks and their execution states for Devr.AI';

3.4 Pydantic Model for Job Declarations

To expose job states on our admin dashboards or monitor them programmatically, we declare a Pydantic representation:

from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime

class ScheduledJobResponse(BaseModel):
    """Schema representing a registered background task in Devr.AI."""
    id: str = Field(..., description="Unique identifier of the background job")
    name: str = Field(..., description="Human-readable name of the function being executed")
    next_run_time: Optional[datetime] = Field(None, description="Timestamp of the next planned execution")
    trigger: str = Field(..., description="Type of trigger (interval, cron, or date)")
    coalesce: bool = Field(True, description="Whether to merge multiple missed executions into a single run")
    is_paused: bool = Field(False, description="Indicates if the job is temporarily suspended")

3.5 Python Implementation: Centralized Scheduler Service

We propose adding a centralized scheduler module under backend/app/core/scheduler.py:

import logging
from apscheduler.schedulers.asyncio import AsyncScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from app.core.config import settings
from app.database.supabase.client import get_supabase_client

logger = logging.getLogger(__name__)

# Centralized Scheduler Instance
scheduler: Optional[AsyncScheduler] = None

def get_scheduler() -> AsyncScheduler:
    global scheduler
    if scheduler is None:
        # Use our existing Supabase/PostgreSQL connection string
        database_url = settings.database_url.replace("postgres://", "postgresql+psycopg2://")
        
        job_stores = {
            'default': SQLAlchemyJobStore(url=database_url, tablename="apscheduler_jobs")
        }
        
        job_defaults = {
            'coalesce': True,
            'max_instances': 1,
            'misfire_grace_time': 300
        }
        
        scheduler = AsyncScheduler(
            jobstores=job_stores,
            job_defaults=job_defaults,
            timezone="UTC"
        )
    return scheduler


# Example Scheduled Task: Periodic Repository Stats Sync
async def sync_repository_stats_job():
    """Fetches up-to-date star/fork/issue metrics from GitHub and updates PostgreSQL."""
    logger.info("Starting scheduled job: sync_repository_stats_job")
    try:
        supabase = get_supabase_client()
        # Fetch active repositories
        response = await supabase.table("repositories").select("id, full_name").execute()
        
        if not response.data:
            logger.info("No repositories to sync.")
            return

        for repo in response.data:
            repo_id = repo['id']
            full_name = repo['full_name']
            logger.debug(f"Syncing stats for {full_name}")
            
            # 1. Fetch latest stats via GitHub Service (simulated/imported)
            # star_count, fork_count, open_issues = await github_service.fetch_repo_stats(full_name)
            
            # 2. Update database
            # await supabase.table("repositories").update({
            --     "stars_count": star_count,
            --     "forks_count": fork_count,
            --     "open_issues_count": open_issues,
            --     "updated_at": datetime.utcnow().isoformat()
            -- }).eq("id", repo_id).execute()
            
        logger.info("Completed scheduled job: sync_repository_stats_job successfully")
    except Exception as e:
        logger.error(f"Error executing sync_repository_stats_job: {e}", exc_info=True)


def register_standard_jobs(sched: AsyncScheduler):
    """Registers standard, recurring system-wide jobs."""
    
    # Sync GitHub metrics every 6 hours
    sched.add_job(
        sync_repository_stats_job,
        trigger='interval',
        hours=6,
        id='sync_repository_stats',
        replace_existing=True
    )
    
    # Example: Run conversation summarization compile daily at midnight UTC
    # sched.add_job(
    #     compile_conversation_summaries_job,
    #     trigger='cron',
    #     hour=0,
    #     minute=0,
    #     id='daily_conversation_compile',
    #     replace_existing=True
    # )

FastAPI Lifespan Integration (backend/app/main.py):

To start and stop the scheduler automatically with our API server:

from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.core.scheduler import get_scheduler, register_standard_jobs

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    logger.info("Initializing system scheduler...")
    sched = get_scheduler()
    register_standard_jobs(sched)
    sched.start()
    logger.info("Background scheduler started successfully.")
    
    yield
    
    # Shutdown
    logger.info("Shutting down background scheduler...")
    sched.shutdown()
    logger.info("Scheduler shutdown complete.")

app = FastAPI(lifespan=lifespan)

---

4. Implementation Steps & Feedback

We request the community's and maintainers' feedback on:

1. Index Coverage: Are there other high-frequency lookups performed by our LangGraph agents (e.g., Weaviate/FalkorDB sync scripts) that we should include in the SQL index migration?
2. Scheduler Store: Is SQLAlchemy/PostgreSQL preferred for the persistent Job Store, or should we consider Redis if Devr.AI scales towards a larger event-driven topology? Using Supabase/PostgreSQL is currently the most lightweight path as we already run it.

Upon agreement, we will provide a Pull Request containing:

• Database migration script 02_optimize_indexes_and_jobs.sql.
• Pyproject dependencies update (adding apscheduler and sqlalchemy).
• Implementation of app/core/scheduler.py and integration in the FastAPI lifespan.