[Feature Request] Add a new job system to reitti

[dgraf-gh]

Context & Problem Statement

Currently, Reitti processes data imports (GPX/Google JSON) and API batches by writing directly to our production tables. This presents a few risks:

1. Lack of Resiliency: If the server crashes mid-import, we end up with partial/corrupt data.
2. No Undo: Once an import starts, it is difficult to stop or revert before it affects the user's timeline.
3. Contention: High-volume imports and real-time API points can cause lock contention on the same tables.

Proposed Solution: The "Staging-to-Promotion" Workflow

I will implement a job system using JobRunr (OSS) to manage all data entry points. We will introduce a Staging Table to act as a buffer before data is promoted to the final timeline.

Architectural Overview

The system relies on three pillars:

• Staging Table (staged_gps_points): A lightweight table with minimal indexing to store raw data during the parsing phase.
• JobRunr OSS: The background job engine. It handles scheduling, retries, and persistence using our existing SQL database.
• Manual Sequential Lock: Since JobRunr OSS does not support named Mutexes, I will implement a manual database lock mechanism to ensure only one promotion job runs at a time.

---

The Data Lifecycle

1. Ingestion Phase (Parallel)

• Files: User uploads a file. We stream-parse it into staged_gps_points tagged with a job_id.
• API Batches: The PointCollector accumulates 1,000 points and flushes them into staged_gps_points, immediately triggering a promotion job.

2. The Grace Period (Cancelable)

• For file imports, once parsing is complete, the promotion step is scheduled with a 5-minute delay.
• If a user cancels during this window, the scheduled job is deleted via jobScheduler.delete(jobId) and staging rows are purged.

3. Promotion Phase (Sequential & Resilient)

• The job wakes and attempts to acquire a database-level lock.
• Data is moved via INSERT INTO final_table SELECT ... FROM staged_gps_points WHERE job_id = X.
• This phase is non-cancelable to maintain transactional integrity.

4. Enrichment Phase (Background)

• Upon successful promotion, a continuation job is fired for reverse geocoding. This runs in parallel as it does not block the primary import flow.

---

Job Monitoring and Progress Estimation

Since JobRunr OSS does not provide a native progress tracking field or custom UI, I will implement the following:

• Job Visibility: I will query the StorageProvider directly to retrieve lists of SCHEDULED (pending) and PROCESSING (running) jobs for the Reitti dashboard.
• Progress Tracking: Progress will be calculated by comparing the row count in staged_gps_points for a specific job_id against the total point count recorded during the initial ingestion phase.

---

Implementation Plan

• JobRunr OSS Integration
  • Integrate jobrunr-spring-boot-3-starter.
  • Configure SQL persistence for job metadata.
• Staging & Lock Schema
  • Create staged_gps_points table.
  • Implement a job_locks table or a SELECT FOR UPDATE mechanism for sequential promotion.
• Core Services
  • StagingService: Handle high-speed batch inserts using JdbcTemplate.
  • PromotionService: Implement promotion logic with manual locking.
• API & Monitoring
  • Update PointCollector to hand off batches to the staging pipeline.
  • Build endpoints to query job status and calculate progress percentages.
• Cleanup Worker
  • Implement a task to purge orphaned rows in the staging table from failed or expired uploads.

---

Example Implementation Sketch

// Scheduling an import with a 5-minute grace period
public void startImport(InputStream stream, Long userId) {
    String jobId = jobScheduler.enqueue(() -> stagingService.parse(stream, userId));
    
    jobScheduler.schedule(
        Instant.now().plus(5, ChronoUnit.MINUTES),
        () -> promotionService.promote(jobId)
    );
}

// Sequential promotion using a manual lock
public void promote(String jobId) {
    lockService.executeWithLock("IMPORT_LOCK", () -> {
        promotionService.moveDataToFinal(jobId);
        jobScheduler.enqueue(() -> enrichmentService.process(jobId));
    });
}

// Fetching pending jobs for the UI
public List<JobRecord> getPendingJobs() {
    return storageProvider.getJobs(StateName.SCHEDULED, PageRequest.ascOnUpdatedAt(100));
}

---

Why this is a win for Reitti

• Stability: JobRunr ensures that no imports are lost if the application restarts.
• Scalability: The staging table prevents heavy write operations from affecting production table performance.
• Data Integrity: The sequential lock and staging pattern prevent partial imports and database bloat.
• User Control: The grace period allows for easy cancellation of accidental uploads.

[dgraf-gh]

We are using dbScheduler now. JobRunnr was a dead end since they only offer the oss version to promote the paid pro version and on every corner in the code this is visible