# 1. Judul
**Desain Optimasi Performa Ingestion (Concurrency, Batching, & Direct DB)**

# 2. Tanggal
4 Mei 2026

# 3. Deskripsi
Dokumen ini menguraikan rencana teknis untuk mengatasi *bottleneck* performa pada proses *ingestion* dan *edge generation* di dalam layanan `app/services/ingestion.py`. Fokus optimasi meliputi migrasi penyimpanan *chunk* Supabase dari REST API yang lambat (karena autentikasi bermasalah/401) ke *Direct PostgreSQL*, serta menekan durasi tunggu LLM (*DashScope*) dengan implementasi *Prompt Batching*, *Multi-Threading* pada *worker*, dan penambahan mekanisme *Retry with Exponential Backoff*.

# 4. Metodologi
Peningkatan performa menggunakan beberapa pola utama:
*   **Direct Database Access**: Menghilangkan lapisan *overhead* jaringan REST API dan menggunakan *driver* `psycopg` untuk *bulk insert* langsung ke PostgreSQL.
*   **Prompt Batching (LLM Optimization)**: Menggabungkan beberapa permintaan pembuatan *reasoning* pilar ke dalam 1 panggilan API JSON untuk setiap pasangan dokumen (A-B). Ini mengubah *N calls per pair* menjadi *1 call per pair*.
*   **Concurrency (Threading)**: Memanfaatkan *ThreadPoolExecutor* untuk menangani beberapa pasangan dokumen secara paralel, mencegah satu proses analisis memblokir antrean secara keseluruhan.
*   **Defensive Programming**: Menambahkan *retry* otomatis dengan jeda eksponensial pada panggilan LLM, mencegah sistem gagal total jika API mengalami *timeout* sesaat.

# 5. Keputusan (Decision Log)
1.  **Hapus Logika REST Supabase**: Modul `app/services/storage/supabase_client.py` akan dimodifikasi secara drastis untuk **hanya** menggunakan koneksi *Direct PostgreSQL*. Fitur REST API Supabase dimatikan karena memakan *overhead* saat *fallback*.
2.  **Prompt Batching per Pasangan**: Saat dua jurnal dinyatakan mirip pada 2 atau lebih pilar (misal: Gap dan Objective), LLM akan diminta untuk membuat *reasoning* keduanya dalam satu *call* yang mengembalikan JSON `{ "gap": "...", "objective": "..." }`.
3.  **Tingkatkan Worker Thread**: Mengganti implementasi *single worker thread* di `ingestion.py` dengan *ThreadPoolExecutor* berkapasitas 3-5 *workers*.
4.  **Batch Neo4j UNWIND**: Pembuatan relasi kemiripan (*edge*) akan disatukan menggunakan transaksi berjenjang (`UNWIND`) di Neo4j untuk menghemat *round-trip time*.
5.  **Persiapan OpenAI Embedding**: Menyiapkan struktur variabel atau blok kondisi (*stub*) untuk model `text-embedding-3-small` (dengan parameter `dimensions=768`), namun mempertahankan model Gemini sebagai *default* yang aktif saat ini.
6.  **Sistem Retry LLM**: Semua pemanggilan LLM (*extraction* & *reasoning*) dibungkus dalam *try-except loop* dengan maksimal 3 kali retri.

# 6. Alur Diagram
```mermaid
flowchart TD
    classDef main fill:#f9f9f9,stroke:#0071C5,stroke-width:2px,color:#232F3E
    classDef worker fill:#fff,stroke:#FF9900,stroke-width:2px,color:#232F3E
    classDef db fill:#eaffea,stroke:#1e8900,stroke-width:2px,color:#1e8900,shape:cylinder

    subgraph Orchestrator ["Ingestion Service"]
        direction TB
        Node[Neo4j MERGE Node]:::main
        Node --> Queue["Task Queue (Node IDs)"]:::main
        
        subgraph Workers ["ThreadPool (3-5 Workers)"]
            direction TB
            W1["Worker 1 (A vs B)"]:::worker
            W2["Worker 2 (A vs C)"]:::worker
            W3["Worker 3 (D vs E)"]:::worker
        end
        Queue --> Workers
        
        W1 --> |"1 LLM Call (Batch: Gap & Method)"| LLM["DashScope / Qwen API<br>(JSON Response + Retry)"]:::worker
        W2 --> |"1 LLM Call (Batch: All Pillars)"| LLM
        W3 --> |"1 LLM Call (Batch: Background)"| LLM
        
        LLM --> |Return Batched JSON| EdgeBatch["Neo4j Edge Batch (UNWIND)"]:::main
    end

    EdgeBatch -.-> Neo4j[("Neo4j AuraDB")]:::db
```

# 7. Struktur Folder (Tree)
File yang terdampak dalam fase ini adalah:
```text
poc_neo4j_migrating-chunking-integrated-gh-(Restruct)/
├── app/
│   └── services/
│       ├── ingestion.py                 # Mengalami refactor besar (Threading & Batching)
│       └── storage/
│           └── supabase_client.py       # Refactor menjadi Direct PostgreSQL Only
└── .env                                 # Penambahan (opsional) OPENAI_API_KEY
```

# 8. Asumsi
1.  *Database* Neo4j AuraDB mampu menerima 3 hingga 5 koneksi simultan (*concurrent transactions*) yang dihasilkan oleh *ThreadPoolExecutor*.
2.  Variabel `DATABASE_URL` atau `DIRECT_URL` di dalam `.env` berisi *connection string* PostgreSQL yang valid.
3.  Pustaka `psycopg2` atau `psycopg` sudah terpasang dan berfungsi (*requirements.txt* sudah memadai).
4.  LLM Qwen via DashScope memiliki instruksi yang cukup kuat untuk selalu membalas dengan struktur JSON murni tanpa pembungkus Markdown saat diminta.

# 9. Konsep Tabel Metriks
Matriks perbandingan target optimalisasi:

| Metrik | Cara Lama (Before) | Cara Baru (After) | Harapan Efisiensi |
| :--- | :--- | :--- | :--- |
| **Supabase Chunk Insert** | Coba REST (Gagal 401, tunggu 5 detik) -> Fallback Postgres | Langsung Eksekusi PostgreSQL via `psycopg` | Menghemat ~5 detik per *batch*. |
| **LLM Call per Candidate Pair** | $N$ panggilan (misal 3 pilar = 3 kali API Call) | 1 panggilan (*JSON Batched Response*) | Hemat hingga 300% durasi & *bandwidth*. |
| **Edge Worker Concurrency** | 1 Dokumen memblokir dokumen lain di antrean (*Single Thread*) | 3-5 Dokumen diproses bersamaan (*Thread Pool*) | Pengurangan total waktu *Ingestion* hingga 70%. |
| **LLM Resilience** | *Timeout* = *Edge/Extraction* Gagal | *Timeout* = Tunggu (*backoff*) & *Retry* | Mencegah *data loss* akibat fluktuasi API. |

# 10. Konsep Logika Prompting
Daripada mengirim 3 *prompt* untuk pilar yang berbeda, kita menyusun 1 *prompt* terpadu:

```text
Task: 
Buat reasoning kemiripan edge untuk pilar-pilar berikut: [Background, Gap, Objective].

Source title: {source_title}
Target title: {target_title}

Data Pilar:
- [BACKGROUND]
  Source: {source_bg_text}
  Target: {target_bg_text}
- [GAP]
  Source: {source_gap_text}
  ...

Aturan:
1. ...
2. Return ONLY a valid JSON object map:
{{
  "background": "reasoning...",
  "gap": "reasoning...",
  "objective": "reasoning..."
}}
```

# 11. Langkah Implementasi
1.  **Refactor `supabase_client.py`**: Buang instansiasi REST API Supabase. Gunakan murni fungsi PostgreSQL untuk `check_supabase_hash` dan `insert_chunks_batch`.
2.  **Siapkan Stub OpenAI Embedding**: Tambahkan inisiasi dan blok kondisi pemanggilan `embed_text` di `ingestion.py` untuk menyuntikkan OpenAI (*Dimension* 768) jika flag tertentu diaktifkan, walau Gemini tetap jadi *default*.
3.  **Terapkan Retry Mechanisms**: Bungkus `extract_pillars_with_llm` dan `embed_text` dengan `for attempt in range(max_retries)` menggunakan `time.sleep` bertahap (*exponential backoff*).
4.  **Refactor Batched Edge Reasoning**: Ubah `generate_edge_reasoning` agar menerima daftar (*list*) pilar yang mirip dari satu kandidat, lalu menyusunnya menjadi 1 *prompt JSON-based*. Tangani respons JSON dari model untuk diekstrak menjadi kamus (`dict`) *reasoning*.
5.  **Multi-Threaded Worker**: Ubah `worker_edge_generator` agar berjalan menggunakan `concurrent.futures.ThreadPoolExecutor` ketimbang satu blok `while True` tunggal yang melayani antrean secara serial. Gabungkan penyimpanan Neo4j menggunakan *batch transaction* (`UNWIND`).

# 12. Tabel Eksekusi Task

| Check Box | Fase | Task | Detail Task | Deskripsi | Acceptance Criteria |
| :---: | :---: | :---: | :--- | :--- | :--- |
| [x] | 1 | DB Refactor | **Direct PostgreSQL Supabase** | Hapus logika inisiasi *Client* Supabase REST di `app/services/storage/supabase_client.py`. Paksa fungsi `check` dan `insert` untuk menggunakan skema *Direct DB* `psycopg`. | Tidak ada pesan log *"Supabase REST insert failed... Falling back"* saat eksekusi. *Insert* langsung berhasil. |
| [x] | 1 | Preparation | **OpenAI Embedding Stub** | Impor *library* `openai` di `ingestion.py`, buat fungsi pembungkus *embedding* untuk menggunakan `text-embedding-3-small` (dengan `dimensions=768`) yang dapat ditukar secara fleksibel dari model Gemini. | Kode `embed_text` memiliki dua cabang pemrosesan (Gemini dan OpenAI) tanpa merusak fungsi Gemini saat ini. |
| [x] | 2 | LLM Refactor | **Prompt Batching Edge Reasoning** | Ubah fungsi `generate_edge_reasoning` untuk menyatukan semua teks pilar dari satu target jurnal dalam 1 *prompt* dan mengembalikan *Dictionary JSON* untuk setiap pilar yang valid. | LLM hanya dipanggil 1 kali per pasangan jurnal, dan respons dapat di-_parse_ menjadi `dict` (atau gagal dengan aman). |
| [x] | 2 | LLM Refactor | **LLM Retry Mechanism** | Bungkus proses panggilan ekstraksi 5 pilar dan pembuatan *reasoning* di `ingestion.py` dengan logika *try-except* (maksimal 3 *retry*) dan *exponential backoff*. | Jika API DashScope atau Gemini gagal/*timeout*, sistem mencoba lagi sebelum mencetak *error* menyerah. |
| [x] | 3 | Concurrency | **ThreadPool Worker** | Tulis ulang eksekutor pada `worker_edge_generator` dengan menggunakan antrean berbasis `ThreadPoolExecutor` yang memiliki batas 3-5 pekerja serentak. | Terminal mencetak proses log pembandingan kemiripan jurnal secara paralel/berdekatan untuk jurnal yang berbeda. |
| [x] | 3 | Neo4j Refactor | **Edge Batching Transaction** | Setelah semua pasangan lolos *Coherence Gate* dan LLM merespons JSON *reasoning*, kumpulkan data *edge*-nya dan `MERGE` sekaligus dalam 1 fungsi Transaksi menggunakan sintaks `UNWIND`. | Pembuatan garis relasi antar 1 Node baru ke beberapa kandidat dieksekusi dalam 1 kali kueri ke Neo4j. |
| [x] | 4 | Validation | **End-to-End Test (Ingestion)** | Kosongkan basis data secara bertahap (atau dengan mode `DRY_RUN_AUDIT=1`) dan eksekusi `ingestion.py` pada seluruh folder PDF yang ada untuk mengukur kestabilan dan kecepatan penyelesaian. | Log tidak mengalami kelambatan *hang* 30 detik berulang kali, tidak ada kegagalan otorisasi Supabase REST, dan sistem sukses berjalan dari ujung ke ujung. |