import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.LocalDate;
import java.time.format.DateTimeParseException;
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;

/** Pipeline penjualan: baca -> validasi -> transformasi -> agregasi -> tulis. */
public final class Pipeline {

    /** Kategori yang dikenal validator; di luar daftar ini baris ditolak. */
    private static final Set<String> KATEGORI_DIKENAL =
            Set.of("Elektronik", "Pakaian", "Buku", "Peralatan");

    /** Faktor biaya layanan 2% yang dihitung tahap transformasi. */
    private static final BigDecimal BIAYA_LAYANAN = new BigDecimal("1.02");

    private Pipeline() {
        // Kelas utilitas: seluruh anggota static.
    }

    /** Tahap baca: memberi nomor urut pada tiap baris berkas. */
    public static List<BarisData> baca(List<String> isiBerkas) {
        List<BarisData> baris = new ArrayList<>(isiBerkas.size());
        for (int i = 0; i < isiBerkas.size(); i++) {
            baris.add(new BarisData(i + 1, isiBerkas.get(i)));
        }
        return baris;
    }

    /** Tahap validasi lalu transformasi untuk satu baris. */
    public static HasilBaris prosesSatuBaris(BarisData baris) {
        String[] segmen = baris.teks().split(";", -1);
        if (segmen.length != 5) {
            return new HasilBaris.TidakSah(baris.nomor(),
                    "jumlah segmen " + segmen.length + ", harus 5");
        }
        LocalDate tanggal;
        try {
            tanggal = LocalDate.parse(segmen[0]);
        } catch (DateTimeParseException e) {
            return new HasilBaris.TidakSah(baris.nomor(),
                    "tanggal tidak terurai: '" + segmen[0] + "'");
        }
        String kota = segmen[1].trim();
        String pelanggan = segmen[2].trim();
        String kategori = segmen[3].trim();
        if (kota.isEmpty()) {
            return new HasilBaris.TidakSah(baris.nomor(), "kota kosong");
        }
        if (pelanggan.isEmpty()) {
            return new HasilBaris.TidakSah(baris.nomor(), "nama pelanggan kosong");
        }
        if (!KATEGORI_DIKENAL.contains(kategori)) {
            return new HasilBaris.TidakSah(baris.nomor(),
                    "kategori tidak dikenal: '" + kategori + "'");
        }
        BigDecimal nilai;
        try {
            nilai = new BigDecimal(segmen[4].trim());
        } catch (NumberFormatException e) {
            return new HasilBaris.TidakSah(baris.nomor(),
                    "nilai tidak terurai: '" + segmen[4].trim() + "'");
        }
        if (nilai.signum() <= 0) {
            return new HasilBaris.TidakSah(baris.nomor(),
                    "nilai harus positif: " + nilai.toPlainString());
        }
        EventPenjualan event = new EventPenjualan(baris.nomor(), tanggal, kota,
                pelanggan, kategori, nilai);
        return new HasilBaris.Sah(ubah(event));
    }

    /** Tahap transformasi: bakukan kota dan hitung nilai berjasa 2%. */
    public static KejadianTerproses ubah(EventPenjualan event) {
        BigDecimal berjasa = event.nilai().multiply(BIAYA_LAYANAN)
                .setScale(2, RoundingMode.HALF_EVEN);
        return new KejadianTerproses(event.nomorBaris(), event.tanggal(),
                event.kota().toUpperCase(Locale.ROOT), event.pelanggan(),
                event.kategori(), event.nilai(), berjasa);
    }

    /** Tahap agregasi: ringkasan terurut dari seluruh kejadian. */
    public static RingkasanAgregasi agregasi(List<KejadianTerproses> kejadian) {
        Map<String, BigDecimal> perKota = kejadian.stream()
                .collect(Collectors.groupingBy(KejadianTerproses::kota,
                        TreeMap::new,
                        Collectors.reducing(BigDecimal.ZERO,
                                KejadianTerproses::nilaiBerjasa, BigDecimal::add)));
        Map<String, Long> perKategori = kejadian.stream()
                .collect(Collectors.groupingBy(KejadianTerproses::kategori,
                        TreeMap::new, Collectors.counting()));
        BigDecimal totalNilai = kejadian.stream()
                .map(KejadianTerproses::nilai)
                .reduce(BigDecimal.ZERO, BigDecimal::add);
        BigDecimal totalBerjasa = kejadian.stream()
                .map(KejadianTerproses::nilaiBerjasa)
                .reduce(BigDecimal.ZERO, BigDecimal::add);
        return new RingkasanAgregasi(kejadian.size(), totalNilai, totalBerjasa,
                perKota, perKategori);
    }

    /** Tahap tulis: susun laporan akhir dari ringkasan dan hasil per baris. */
    public static List<String> tulis(RingkasanAgregasi ringkasan, List<HasilBaris> hasil) {
        int sah = 0;
        for (HasilBaris h : hasil) {
            if (h instanceof HasilBaris.Sah) {
                sah++;
            }
        }
        List<String> laporan = new ArrayList<>();
        laporan.add("=== Laporan pipeline penjualan ===");
        laporan.add("baris dibaca  : " + hasil.size());
        laporan.add("baris sah     : " + sah);
        laporan.add("baris ditolak : " + (hasil.size() - sah));
        laporan.add(String.format(Locale.US, "total nilai   : %,14.2f", ringkasan.totalNilai()));
        laporan.add(String.format(Locale.US, "total berjasa : %,14.2f", ringkasan.totalBerjasa()));
        laporan.add("nilai berjasa per kota:");
        ringkasan.nilaiBerjasaPerKota().forEach((kota, nilai) ->
                laporan.add(String.format(Locale.US, "  %-12s %,14.2f", kota, nilai)));
        laporan.add("jumlah transaksi per kategori:");
        ringkasan.jumlahPerKategori().forEach((kategori, jumlah) ->
                laporan.add(String.format(Locale.US, "  %-12s %6d", kategori, jumlah)));
        laporan.add("baris ditolak (terurut nomor):");
        for (HasilBaris h : hasil) {
            if (h instanceof HasilBaris.TidakSah(int nomor, String alasan)) {
                laporan.add(String.format(Locale.US, "  baris %2d: %s", nomor, alasan));
            }
        }
        return laporan;
    }

    /** Menjalankan seluruh tahapan secara sekuensial. */
    public static HasilPipeline jalankanSekuensial(List<String> isiBerkas) {
        List<BarisData> baris = baca(isiBerkas);
        List<HasilBaris> hasil = new ArrayList<>(baris.size());
        for (BarisData b : baris) {
            hasil.add(prosesSatuBaris(b));
        }
        return rakitHasil(baris.size(), hasil);
    }

    /**
     * Menjalankan tahap validasi + transformasi secara paralel: satu task per
     * baris di executor apa pun. Hasil dikumpulkan dulu, digabung, baru diurutkan.
     */
    public static HasilPipeline jalankanParalel(List<String> isiBerkas, ExecutorService eksekutor) {
        List<BarisData> baris = baca(isiBerkas);
        // Peta konkuren berkunci nomor baris: thread mana pun boleh menulis
        // bersamaan karena kunci tiap baris unik.
        Map<Integer, HasilBaris> hasilPerNomor = new ConcurrentHashMap<>();
        List<CompletableFuture<Void>> tugas = new ArrayList<>(baris.size());
        for (BarisData b : baris) {
            tugas.add(CompletableFuture.runAsync(() ->
                    hasilPerNomor.put(b.nomor(), prosesSatuBaris(b)), eksekutor));
        }
        // Gerbang tunggu tunggal: seluruh task selesai sebelum hasil dibaca.
        CompletableFuture.allOf(tugas.toArray(new CompletableFuture<?>[0])).join();
        // Pulihkan urutan nomor baris lewat salinan TreeMap.
        List<HasilBaris> terurut = new ArrayList<>(new TreeMap<>(hasilPerNomor).values());
        return rakitHasil(baris.size(), terurut);
    }

    /** Merakit hasil per baris menjadi ringkasan, laporan, dan statistik pipeline. */
    private static HasilPipeline rakitHasil(int jumlahBaris, List<HasilBaris> hasil) {
        List<KejadianTerproses> kejadian = new ArrayList<>(hasil.size());
        for (HasilBaris h : hasil) {
            if (h instanceof HasilBaris.Sah(var k)) {
                kejadian.add(k);
            }
        }
        RingkasanAgregasi ringkasan = agregasi(kejadian);
        return new HasilPipeline(jumlahBaris, kejadian.size(),
                jumlahBaris - kejadian.size(),
                ringkasan.nilaiBerjasaPerKota().size(),
                ringkasan.jumlahPerKategori().size(),
                tulis(ringkasan, hasil));
    }
}