Sangeetha-Grantha

Metadata Value
Status Archived
Version 1.0.0
Last Updated 2026-09-10
Author Sangeetha Grantha Team
Document Type Archive

Bulk Import Orchestration & Ops Plan

[!NOTE] Historical evidence: results, counts, commands, and observations below belong to the original work described here. The editorial update date is not a new test or corpus verification. For present behavior, use current ingestion guide.

Goal

Provide a resilient, observable orchestration for Krithi bulk import starting from CSV manifests, with full progress tracking, failure/retry controls, and an admin-facing dashboard—built using existing stack (Kotlin Ktor backend, React admin web, Postgres).

Recommendation (Best-Fit Option)

Architecture Rationale

Why NOT sangita-cli for Production Operations?

sangita-cli is developer tooling, not production infrastructure:

  1. Purpose: Designed for local development workflows:
    • Database management (db reset, db migrate)
    • Development server orchestration (dev --start-db)
    • Testing (test steel-thread)
    • Git guardrails (commit check)
    • Network/mobile setup utilities
  2. Architectural Mismatch:
    • CLI tools are meant for one-off operations, not long-running services
    • Would require subprocess spawning from backend (complex error handling, resource management)
    • No native integration with Ktor’s coroutine ecosystem
    • Would mix dev tooling concerns with production business logic
  3. Operational Concerns:
    • CLI processes are harder to monitor, scale, and restart
    • No built-in health checks or graceful shutdown
    • Would require separate deployment/runtime management
    • Breaks the service-oriented architecture pattern

Why Ktor Backend with Coroutines?

Kotlin coroutines provide native async processing within the existing backend:

  1. Architectural Alignment:
    • All existing services already use suspend functions
    • WebScrapingService and ImportService are already in the backend
    • Consistent error handling, logging, and audit patterns
    • Single codebase, single deployment unit
  2. Technical Advantages:
    • CoroutineScope: Built-in structured concurrency for background jobs
    • Channels/Flow: Natural task queue patterns
    • SupervisorJob: Isolated failure handling per batch
    • CoroutineContext: Easy cancellation, timeouts, and resource management
    • Integration: Direct access to DatabaseFactory, SangitaDal, and existing services
  3. Operational Benefits:
    • Single process to monitor and scale
    • Unified logging and metrics
    • Graceful shutdown via Ktor lifecycle hooks
    • Health checks via existing /health endpoint
    • No subprocess management overhead

Architecture Overview

Components

Backend (Ktor) - Core Orchestration

Service Layer:

API Routes (/v1/admin/bulk-import):

Background Processing:

Lifecycle Management:

Frontend (React Admin Web)

Dashboard Components:

Proposed Data Model (Postgres)

Orchestration Flow (Happy Path)

  1. Batch Create (Admin API):
    • POST /v1/admin/bulk-import/batches with CSV file or manifest path
    • Backend creates import_batch row with status pending
    • Returns batch ID immediately (async processing)
  2. Manifest Ingest Job (Background Worker):
    • Worker coroutine picks up batch, changes status to running
    • Parses CSV file, validates entries
    • Creates import_job (type: manifest_ingest) and import_task_run rows (one per CSV entry)
    • Updates batch totals (total_tasks, processed_tasks)
    • Marks job as succeeded, enqueues scrape tasks
  3. Scrape/Enrich Jobs (Background Workers):
    • Multiple worker coroutines poll for pending tasks with job_type = 'scrape'
    • Each worker:
      • Fetches URL using WebScrapingService.scrapeKrithi()
      • Parses and normalizes metadata
      • Creates staging record via ImportService.submitImports()
      • Updates task status to succeeded or failed
      • Updates batch progress counters
  4. Entity Resolution Job (Background Worker):
    • Worker processes tasks with job_type = 'entity_resolution'
    • Calls existing resolution logic to map composer/raga/temple/etc.
    • Stores confidence scores in task result JSONB
    • Marks low-confidence tasks as blocked for manual review
  5. Review Prep Job (Background Worker):
    • Worker processes tasks ready for review
    • Attaches evidence links, aggregates metadata
    • Marks tasks as ready_for_review
    • Updates batch status when all tasks complete
  6. Batch Completion:
    • When all tasks reach terminal status (succeeded, failed, blocked, cancelled)
    • Batch status updated to succeeded (if any succeeded) or failed (if all failed)
    • Final metrics calculated and stored

Failure, Retry, and Idempotency

Dashboard & Ops UX

Security & Governance

Implementation Details

Coroutine-Based Worker Pattern

class BulkImportWorkerService(
    private val dal: SangitaDal,
    private val importService: ImportService,
    private val webScrapingService: WebScrapingService
) {
    private val workerScope = CoroutineScope(
        SupervisorJob() + Dispatchers.IO + 
        CoroutineName("BulkImportWorkers")
    )
    
    fun startWorkers(config: WorkerConfig) {
        repeat(config.scrapeWorkerCount) {
            workerScope.launch {
                processScrapeTasks()
            }
        }
        // ... other worker types
    }
    
    private suspend fun processScrapeTasks() {
        while (isActive) {
            val task = dal.bulkImport.getNextPendingTask("scrape")
            if (task != null) {
                try {
                    processScrapeTask(task)
                } catch (e: Exception) {
                    handleTaskFailure(task, e)
                }
            } else {
                delay(1000) // Poll interval
            }
        }
    }
}

Integration with Existing Services

Configuration

data class BulkImportConfig(
    val scrapeWorkerCount: Int = 3,
    val maxConcurrentScrapes: Int = 5,
    val scrapeRateLimitPerDomain: Int = 10, // per minute
    val taskPollInterval: Duration = Duration.ofSeconds(1),
    val stuckTaskThreshold: Duration = Duration.ofMinutes(30),
    val maxRetries: Int = 3,
    val retryBackoffBase: Duration = Duration.ofSeconds(5)
)

Rollout Plan (Incremental)

  1. Phase A (Foundation):
    • Create Postgres tables (import_batch, import_job, import_task_run, import_event)
    • Implement BulkImportOrchestrationService with batch lifecycle APIs
    • Implement CSV parsing and manifest ingestion worker
    • Basic dashboard: batch list + batch detail with task table
  2. Phase B (Scrape & Enrich):
    • Implement scrape worker coroutines with rate limiting
    • Integrate with existing WebScrapingService and ImportService
    • Dashboard: error drill-down, retry controls, progress indicators
  3. Phase C (Entity Resolution):
    • Implement entity resolution worker
    • Add confidence scoring and blocked state for low-confidence matches
    • Dashboard: confidence filters, resolution suggestions
  4. Phase D (Review Workflow):
    • Frontend review queue integration
    • Approvals push to main import pipeline via existing ImportService.reviewImport()
    • Export functionality for QA (CSV error reports)
  5. Phase E (Hardening):
    • Stuck task detector (coroutine-based scheduler)
    • SLO monitoring and alerts
    • Performance tuning (worker pool sizing, batch size optimization)
    • Load test on ~1,240 entries from CSV files

Alternatives Considered

1. External Orchestrators (Temporal/Airflow/Prefect)

2. Koog (Kotlin-based Pipeline Framework)

3. Rust sangita-cli Workers

4. Pure SQL COPY

5. Ktor Backend with Coroutines ✅ SELECTED

Technical Considerations

Worker Lifecycle Management

Startup (in App.kt):

val bulkImportService = BulkImportOrchestrationService(dal, importService, webScrapingService)
val workerService = BulkImportWorkerService(dal, importService, webScrapingService)

embeddedServer(Netty, host = env.host, port = env.port) {
    // ... existing configuration
    
    // Start background workers
    workerService.startWorkers(config)
    
    monitor.subscribe(ApplicationStopping) {
        logger.info("Shutting down bulk import workers")
        workerService.stopWorkers() // Graceful cancellation
        DatabaseFactory.close()
    }
}.start(wait = true)

Task Queue Pattern

Error Handling & Retries

Actionable Next Steps

  1. Approve Architecture: Review and approve this Ktor-based approach
  2. Database Schema: Create migration for orchestration tables
  3. Phase A Implementation:
    • Implement BulkImportOrchestrationService and BulkImportWorkerService
    • Create batch/job/task APIs
    • Implement CSV manifest ingestion worker
    • Build initial dashboard (batch list + detail)
  4. Testing: Unit tests for services, integration tests for full workflow
  5. Documentation: Update API contract and architecture docs

Documentation home · Feature status