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

# RFC: Database Index Schema Optimizations & Standardized Asynchronous Task Scheduling

## Status: Draft
**Author:** Antigravity (AI Contributor)
**Target Repository:** [AOSSIE-Org/Devr.AI](https://github.com/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](file:///Users/ahmadfaraz/Codes/GSoC/repos/aossie/Devr.AI/backend/app/database/supabase/scripts/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:

```sql
-- ====================================================================
-- 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
```sql
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')
```sql
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.

```mermaid
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]
```

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

```sql
-- 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:

```python
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`:

```python
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:

```python
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.

Start by reading backend/app/database/supabase/scripts/create_db.sql and locating the FastAPI lifespan entry point and existing cron or background-task code. Compare those entry points with the proposed backend/app/core/scheduler.py design, PostgreSQL job store, and listed scheduled jobs. Done requires an agreed migration and scheduler implementation covering the stated indexing and task-scheduling requirements.

索引モデルが issue の本文から書いたものです。