Bagian 3: Integrasi PostgreSQL

[package]
name = "api-katalog"
version = "1.0.0"
edition = "2021"
description = "REST API katalog item dengan Rust, axum, dan PostgreSQL"
license = "MIT"

[dependencies]
# Runtime async standar de facto.
tokio = { version = "1", features = ["full"] }
# Web framework ringan berbasis tower.
axum = "0.8"
# Driver PostgreSQL async.
tokio-postgres = { version = "0.7", features = ["with-serde_json-1"] }
# Pool koneksi di atas tokio-postgres.
deadpool-postgres = "0.14"
# Serialisasi JSON untuk body request dan respons.
serde = { version = "1", features = ["derive"] }
serde_json = "1"

[dev-dependencies]
# HTTP client untuk integration test black-box. default-features
# dimatikan agar tidak menarik native-tls (butuh OpenSSL): test hanya
# menembak http:// sehingga TLS tidak diperlukan.
reqwest = { version = "0.12", default-features = false, features = ["json"] }
# Fitur macros + rt-multi-thread untuk #[tokio::test] di tests/.
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }

[profile.release]
lto = true
strip = true
opt-level = "s"

name: katalog-rust

services:
  app:
    build: .
    image: katalog-rust-app:1.0
    ports:
      - "8081:8080"
    environment:
      PGHOST: db
      PGPORT: "5432"
      PGUSER: katalog
      PGPASSWORD: ${POSTGRES_PASSWORD:?wajib diisi}
      PGDATABASE: katalog
    depends_on:
      db:
        condition: service_healthy
    restart: unless-stopped

  db:
    image: postgres:18-alpine
    environment:
      POSTGRES_USER: katalog
      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:?wajib diisi}
      POSTGRES_DB: katalog
    volumes:
      - db-data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U katalog -d katalog"]
      interval: 2s
      timeout: 2s
      retries: 15
    restart: unless-stopped

  # Service test hanya berjalan dengan profile khusus:
  #   docker compose --profile test run --rm test
  test:
    profiles: ["test"]
    image: rust:1.99-alpine
    working_dir: /src
    volumes:
      - ./:/src
    environment:
      APP_URL: http://app:8080
      PGHOST: db
      PGPORT: "5432"
      PGUSER: katalog
      PGPASSWORD: ${POSTGRES_PASSWORD:?wajib diisi}
      PGDATABASE: katalog
    depends_on:
      app:
        condition: service_started
      db:
        condition: service_healthy
    command: cargo test

volumes:
  db-data:

-- Migrasi pertama: tabel utama items.
CREATE TABLE IF NOT EXISTS items (
    id         BIGSERIAL PRIMARY KEY,
    name       TEXT NOT NULL,
    note       TEXT NOT NULL DEFAULT '',
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

//! Aplikasi katalog: REST API sederhana dengan axum yang menyimpan
//! daftar item di PostgreSQL lewat pool koneksi deadpool-postgres.
//!
//! Konfigurasi dibaca dari environment variable (pola twelve-factor):
//! APP_PORT menentukan port HTTP, MIGRATIONS_DIR lokasi berkas migrasi,
//! dan variabel PG* (PGHOST, PGPORT, PGUSER, PGPASSWORD, PGDATABASE)
//! menentukan koneksi PostgreSQL — sama seperti aplikasi Go pendahulunya.

use std::time::Duration;

use axum::{
    extract::{Path, State},
    http::StatusCode,
    routing::get,
    Json, Router,
};
use deadpool_postgres::{Manager, ManagerConfig, Pool, RecyclingMethod};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use tokio_postgres::NoTls;

/// Kolom item yang dipakai semua query. created_at diformat RFC 3339 UTC
/// lewat to_char agar bentuknya sama seperti time.Time pada versi Go.
const KOLOM_ITEM: &str = "id, name, note, \
     to_char(created_at AT TIME ZONE 'UTC', 'YYYY-MM-DD\"T\"HH24:MI:SS.US\"Z\"')";

/// Batas panjang field note, sama seperti versi Go.
const BATAS_CATATAN: usize = 200;

/// Konfigurasi koneksi database dan server HTTP.
struct Konfigurasi {
    app_port: String,
    migrations_dir: String,
    pg: tokio_postgres::Config,
}

impl Konfigurasi {
    /// Membaca APP_PORT, MIGRATIONS_DIR, dan variabel PG* dari environment.
    /// Nilai default dipilih agar cocok dengan service `db` di compose.yaml.
    fn dari_env() -> Self {
        let ambil = |kunci: &str, standar: &str| {
            std::env::var(kunci).unwrap_or_else(|_| standar.to_string())
        };
        let mut pg = tokio_postgres::Config::new();
        pg.host(&ambil("PGHOST", "db"))
            .port(ambil("PGPORT", "5432").parse().unwrap_or(5432))
            .user(&ambil("PGUSER", "katalog"))
            .password(&ambil("PGPASSWORD", "rahasia123"))
            .dbname(&ambil("PGDATABASE", "katalog"));
        Self {
            app_port: ambil("APP_PORT", "8080"),
            migrations_dir: ambil("MIGRATIONS_DIR", "migrations"),
            pg,
        }
    }
}

/// Item adalah satu baris tabel items; derive Serialize menentukan bentuk JSON.
#[derive(Serialize)]
struct Item {
    id: i64,
    name: String,
    note: String,
    created_at: String,
}

impl Item {
    /// Memetakan satu baris hasil query ke bentuk respons.
    fn dari_baris(baris: &tokio_postgres::Row) -> Self {
        Self {
            id: baris.get(0),
            name: baris.get(1),
            note: baris.get(2),
            created_at: baris.get(3),
        }
    }
}

/// Body pembuatan item baru; note boleh tidak ada (default string kosong).
#[derive(Deserialize)]
struct ItemBaru {
    name: String,
    #[serde(default)]
    note: String,
}

/// Tipe galat handler: status HTTP + body JSON {"error": "..."}.
type GalatApi = (StatusCode, Json<Value>);

/// Menyusun respons error standar — padanan tulisError pada versi Go.
fn tulis_error(status: StatusCode, pesan: &str) -> GalatApi {
    (status, Json(serde_json::json!({ "error": pesan })))
}

#[tokio::main]
async fn main() {
    let cfg = Konfigurasi::dari_env();

    // Pool koneksi dengan reuse agar query tidak membuka koneksi baru
    // setiap saat; kapasitas 10 koneksi seperti versi Go.
    let manager_config = ManagerConfig {
        recycling_method: RecyclingMethod::Fast,
    };
    let manager = Manager::from_config(cfg.pg.clone(), NoTls, manager_config);
    let pool = Pool::builder(manager)
        .max_size(10)
        .build()
        .expect("gagal membangun pool koneksi");

    terapkan_migrasi(&pool, &cfg.migrations_dir)
        .await
        .expect("gagal menjalankan migrasi");

    // Pool ditanam sebagai state; axum meng-clone-nya untuk setiap
    // request sehingga handler cukup meminta State<Pool>.
    let app = Router::new()
        .route("/healthz", get(handle_healthz))
        .route("/api/items", get(handle_list_items).post(handle_create_item))
        .route("/api/items/{id}", get(handle_get_item))
        .with_state(pool);

    let alamat = format!("0.0.0.0:{}", cfg.app_port);
    let listener = tokio::net::TcpListener::bind(&alamat)
        .await
        .expect("gagal mengikat alamat");

    eprintln!("server mendengarkan di :{}", cfg.app_port);

    // Graceful shutdown: saat SIGTERM/SIGINT diterima, axum berhenti
    // menerima koneksi baru lalu menyelesaikan request yang sedang
    // berjalan. select! memberi masa tenggang 10 detik — padanan
    // srv.Shutdown dengan timeout 10 detik pada aplikasi Go.
    tokio::select! {
        hasil = axum::serve(listener, app).with_graceful_shutdown(sinyal_berakhir()) => {
            if let Err(galat) = hasil {
                eprintln!("server berhenti dengan galat: {galat}");
            }
        }
        _ = async {
            sinyal_berakhir().await;
            eprintln!("sinyal berakhir diterima, mematikan server...");
            tokio::time::sleep(Duration::from_secs(10)).await;
        } => {
            eprintln!("masa tenggang 10 detik tercapai, memaksa berhenti");
        }
    }
}

/// Menangkap SIGTERM atau SIGINT (Ctrl-C) untuk shutdown bersih.
async fn sinyal_berakhir() {
    let ctrl_c = async {
        tokio::signal::ctrl_c()
            .await
            .expect("gagal memasang handler Ctrl-C");
    };
    #[cfg(unix)]
    let terminate = async {
        tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
            .expect("gagal memasang handler SIGTERM")
            .recv()
            .await;
    };
    #[cfg(not(unix))]
    let terminate = std::future::pending::<()>();

    tokio::select! {
        _ = ctrl_c => {},
        _ = terminate => {},
    }
}

/// Menerapkan setiap berkas *.sql di direktori migrasi yang belum tercatat
/// di schema_migrations. Satu berkas berjalan dalam satu transaksi bersama
/// pencatatan versinya, sehingga keduanya sukses atau gagal bersama.
async fn terapkan_migrasi(
    pool: &Pool,
    direktori: &str,
) -> Result<(), Box<dyn std::error::Error>> {
    let mut client = pool.get().await?;

    client
        .execute(
            "CREATE TABLE IF NOT EXISTS schema_migrations (
                version    TEXT PRIMARY KEY,
                applied_at TIMESTAMPTZ NOT NULL DEFAULT now()
            )",
            &[],
        )
        .await?;

    let mut berkas: Vec<String> = std::fs::read_dir(direktori)?
        .filter_map(|entri| entri.ok())
        .filter(|entri| entri.path().is_file())
        .map(|entri| entri.file_name().to_string_lossy().into_owned())
        .filter(|nama| nama.ends_with(".sql"))
        .collect();
    berkas.sort();

    for nama in berkas {
        let sudah: bool = client
            .query_one(
                "SELECT EXISTS (SELECT 1 FROM schema_migrations WHERE version = $1)",
                &[&nama],
            )
            .await?
            .get(0);
        if sudah {
            eprintln!("migrasi {nama} sudah diterapkan, dilewati");
            continue;
        }

        let isi = std::fs::read_to_string(format!("{direktori}/{nama}"))?;

        let tx = client.transaction().await?;
        tx.batch_execute(&isi).await?;
        tx.execute("INSERT INTO schema_migrations (version) VALUES ($1)", &[&nama])
            .await?;
        tx.commit().await?;
        eprintln!("migrasi {nama} diterapkan");
    }
    Ok(())
}

/// GET /healthz — memastikan aplikasi DAN database sama-sama hidup.
async fn handle_healthz(State(pool): State<Pool>) -> Result<Json<Value>, GalatApi> {
    let client = pool
        .get()
        .await
        .map_err(|_| tulis_error(StatusCode::SERVICE_UNAVAILABLE, "database tidak merespons"))?;
    client
        .query_one("SELECT 1", &[])
        .await
        .map_err(|_| tulis_error(StatusCode::SERVICE_UNAVAILABLE, "database tidak merespons"))?;
    Ok(Json(serde_json::json!({ "status": "ok" })))
}

/// GET /api/items — seluruh item sebagai array JSON.
async fn handle_list_items(State(pool): State<Pool>) -> Result<Json<Vec<Item>>, GalatApi> {
    let client = pool
        .get()
        .await
        .map_err(|_| tulis_error(StatusCode::INTERNAL_SERVER_ERROR, "gagal membuka koneksi"))?;
    let baris = client
        .query(
            &format!("SELECT {KOLOM_ITEM} FROM items ORDER BY id"),
            &[],
        )
        .await
        .map_err(|_| tulis_error(StatusCode::INTERNAL_SERVER_ERROR, "gagal membaca items"))?;
    Ok(Json(baris.iter().map(Item::dari_baris).collect()))
}

/// POST /api/items — validasi body lalu simpan item baru.
async fn handle_create_item(
    State(pool): State<Pool>,
    Json(body): Json<Value>,
) -> Result<(StatusCode, Json<Item>), GalatApi> {
    // Bentuk body diperiksa manual agar pesan galat tetap konsisten.
    let input: ItemBaru = serde_json::from_value(body)
        .map_err(|_| tulis_error(StatusCode::BAD_REQUEST, "body JSON tidak valid"))?;

    let name = input.name.trim().to_string();
    if name.is_empty() {
        return Err(tulis_error(StatusCode::BAD_REQUEST, "name wajib diisi"));
    }
    if input.note.chars().count() > BATAS_CATATAN {
        return Err(tulis_error(
            StatusCode::BAD_REQUEST,
            "note maksimal 200 karakter",
        ));
    }

    let client = pool
        .get()
        .await
        .map_err(|_| tulis_error(StatusCode::INTERNAL_SERVER_ERROR, "gagal membuka koneksi"))?;
    let baris = client
        .query_one(
            &format!("INSERT INTO items (name, note) VALUES ($1, $2) RETURNING {KOLOM_ITEM}"),
            &[&name, &input.note],
        )
        .await
        .map_err(|_| tulis_error(StatusCode::INTERNAL_SERVER_ERROR, "gagal menyimpan item"))?;
    Ok((StatusCode::CREATED, Json(Item::dari_baris(&baris))))
}

/// GET /api/items/{id} — satu item; 404 bila tidak ditemukan.
async fn handle_get_item(
    State(pool): State<Pool>,
    Path(id): Path<i64>,
) -> Result<Json<Item>, GalatApi> {
    let client = pool
        .get()
        .await
        .map_err(|_| tulis_error(StatusCode::INTERNAL_SERVER_ERROR, "gagal membuka koneksi"))?;
    let baris = client
        .query_opt(
            &format!("SELECT {KOLOM_ITEM} FROM items WHERE id = $1"),
            &[&id],
        )
        .await
        .map_err(|_| tulis_error(StatusCode::INTERNAL_SERVER_ERROR, "gagal membaca item"))?;
    match baris {
        Some(baris) => Ok(Json(Item::dari_baris(&baris))),
        None => Err(tulis_error(StatusCode::NOT_FOUND, "item tidak ditemukan")),
    }
}

//! Integration test black-box untuk API katalog.
//!
//! Test ini tidak memanggil fungsi internal aplikasi; yang diperiksa adalah
//! perilaku HTTP persis seperti yang dirasakan klien sungguhan. Aplikasi
//! harus sudah berjalan di APP_URL (default http://localhost:8080) dengan
//! database yang dapat dijangkau lewat TEST_DATABASE_URL atau variabel PG*.
//! Jalankan selama stack Compose berjalan:
//!
//!     docker compose --profile test run --rm test
//!
//! Hanya `create_list_get_item` yang mereset tabel items: dua test lainnya
//! tidak menulis data sehingga seluruh test aman dijalankan paralel.

use std::time::{Duration, Instant};

use serde_json::{json, Value};

/// Alamat aplikasi yang diuji (bisa ditimpa APP_URL).
fn app_url() -> String {
    std::env::var("APP_URL").unwrap_or_else(|_| "http://localhost:8080".into())
}

/// Menyusun konfigurasi koneksi PostgreSQL untuk reset data:
/// TEST_DATABASE_URL diutamakan, jika kosong variabel PG* dirangkai
/// dengan nilai default lokal.
fn konfigurasi_db() -> tokio_postgres::Config {
    if let Ok(url) = std::env::var("TEST_DATABASE_URL") {
        return url.parse().expect("TEST_DATABASE_URL tidak valid");
    }
    let ambil = |kunci: &str, standar: &str| {
        std::env::var(kunci).unwrap_or_else(|_| standar.to_string())
    };
    let mut cfg = tokio_postgres::Config::new();
    cfg.host(&ambil("PGHOST", "localhost"))
        .port(ambil("PGPORT", "5432").parse().unwrap_or(5432))
        .user(&ambil("PGUSER", "katalog"))
        .password(&ambil("PGPASSWORD", "rahasia123"))
        .dbname(&ambil("PGDATABASE", "katalog"));
    cfg
}

/// Mengosongkan tabel items dan mengembalikan counter id ke 1 sehingga
/// test dimulai dari kondisi yang sama.
async fn reset_items() {
    let (client, koneksi) = konfigurasi_db()
        .connect(tokio_postgres::NoTls)
        .await
        .expect("gagal membuka koneksi database test");
    tokio::spawn(async move {
        if let Err(galat) = koneksi.await {
            eprintln!("koneksi database test terputus: {galat}");
        }
    });
    client
        .execute("TRUNCATE items RESTART IDENTITY", &[])
        .await
        .expect("gagal truncate items");
}

/// Menunggu /healthz siap, paling lama 30 detik — container sering butuh
/// beberapa detik sebelum benar-benar siap menerima request.
async fn tunggu_server() {
    let tenggat = Instant::now() + Duration::from_secs(30);
    while Instant::now() < tenggat {
        if let Ok(resp) = reqwest::get(format!("{}/healthz", app_url())).await {
            if resp.status().is_success() {
                return;
            }
        }
        tokio::time::sleep(Duration::from_millis(500)).await;
    }
    panic!("aplikasi di {} tidak siap setelah 30 detik", app_url());
}

/// Mengirim POST dengan body JSON ke path dan mengembalikan responsnya.
async fn post_json(path: &str, body: Value) -> reqwest::Response {
    reqwest::Client::new()
        .post(format!("{}{}", app_url(), path))
        .json(&body)
        .send()
        .await
        .expect("permintaan POST gagal")
}

#[tokio::test]
async fn healthz_ok() {
    tunggu_server().await;

    let resp = reqwest::get(format!("{}/healthz", app_url()))
        .await
        .expect("permintaan healthz gagal");
    assert_eq!(resp.status(), 200);
    let body: Value = resp.json().await.expect("healthz bukan JSON");
    assert_eq!(body["status"], "ok");
}

#[tokio::test]
async fn create_list_get_item() {
    tunggu_server().await;
    reset_items().await;

    // Buat item baru: id pertama setelah reset harus 1.
    let resp = post_json(
        "/api/items",
        json!({"name": "Kopi Torabika", "note": "untuk pengujian"}),
    )
    .await;
    assert_eq!(resp.status(), 201);
    let dibuat: Value = resp.json().await.expect("create bukan JSON");
    assert_eq!(dibuat["id"], 1);
    assert_eq!(dibuat["name"], "Kopi Torabika");
    assert_eq!(dibuat["note"], "untuk pengujian");
    assert!(dibuat["created_at"].as_str().is_some());

    // Daftar memuat tepat satu item.
    let resp = reqwest::get(format!("{}/api/items", app_url()))
        .await
        .expect("permintaan list gagal");
    assert_eq!(resp.status(), 200);
    let items: Value = resp.json().await.expect("list bukan JSON");
    assert_eq!(items.as_array().map(Vec::len), Some(1));
    assert_eq!(items[0]["name"], "Kopi Torabika");

    // Ambil kembali berdasarkan id.
    let resp = reqwest::get(format!("{}/api/items/1", app_url()))
        .await
        .expect("permintaan get gagal");
    assert_eq!(resp.status(), 200);
    let didapat: Value = resp.json().await.expect("get bukan JSON");
    assert_eq!(didapat["id"], 1);
    assert_eq!(didapat["name"], "Kopi Torabika");
    assert_eq!(didapat["note"], "untuk pengujian");
}

#[tokio::test]
async fn get_item_404() {
    tunggu_server().await;

    let resp = reqwest::get(format!("{}/api/items/999", app_url()))
        .await
        .expect("permintaan get gagal");
    assert_eq!(resp.status(), 404);
    let body: Value = resp.json().await.expect("error bukan JSON");
    assert_eq!(body["error"], "item tidak ditemukan");
}

#[tokio::test]
async fn create_validasi_400() {
    tunggu_server().await;

    // name kosong (hanya spasi) ditolak sebelum menyentuh database.
    let resp = post_json("/api/items", json!({"name": "   ", "note": ""})).await;
    assert_eq!(resp.status(), 400);
    let body: Value = resp.json().await.expect("error bukan JSON");
    assert_eq!(body["error"], "name wajib diisi");

    // note melebihi batas 200 karakter.
    let resp = post_json(
        "/api/items",
        json!({"name": "Kopi Gayo", "note": "x".repeat(201)}),
    )
    .await;
    assert_eq!(resp.status(), 400);
    let body: Value = resp.json().await.expect("error bukan JSON");
    assert_eq!(body["error"], "note maksimal 200 karakter");
}