import java.math.BigDecimal;
import java.math.RoundingMode;
import java.net.ConnectException;
import java.time.LocalDate;
import java.time.format.DateTimeParseException;
import java.util.ArrayList;
import java.util.Comparator;
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.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.LongAdder;
import java.util.stream.Collectors;

/** Pipeline tahan galat: baca -> validasi -> transformasi (dengan retry) -> 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");

    /** Penanda kondisi layanan yang boleh menempel pada baris data. */
    private static final Set<String> PENANDA_DIKENAL =
            Set.of("transien", "timeout", "rusak");

    /** 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 dengan percobaan ulang untuk satu baris.
     * Setiap percobaan gagal tercatat sebagai jejak; hasil akhir tetap nilai data.
     */
    public static HasilBaris prosesDenganRetry(BarisData baris, KebijakanRetry kebijakan,
            ConcurrentLinkedQueue<CatatanPercobaan> jejak, LongAdder totalTunggu) {
        // --- tahap validasi ---
        String[] segmen = baris.teks().split(";", -1);
        if (segmen.length != 5 && segmen.length != 6) {
            return new HasilBaris.TidakSah(baris.nomor(),
                    "jumlah segmen " + segmen.length + ", harus 5 atau 6");
        }
        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());
        }
        String penanda = segmen.length == 6 ? segmen[5].trim() : "";
        if (!penanda.isEmpty() && !PENANDA_DIKENAL.contains(penanda)) {
            return new HasilBaris.TidakSah(baris.nomor(),
                    "penanda tidak dikenal: '" + penanda + "'");
        }
        EventPenjualan event = new EventPenjualan(baris.nomor(), tanggal, kota,
                pelanggan, kategori, nilai, penanda);

        // --- tahap transformasi dengan percobaan ulang ---
        for (int percobaan = 1; percobaan <= kebijakan.percobaanMaks(); percobaan++) {
            try {
                KejadianTerproses kejadian = prosesDenganLayanan(event, percobaan);
                if (percobaan > 1) {
                    jejak.add(new CatatanPercobaan(baris.nomor(), percobaan,
                            "berhasil setelah percobaan ulang"));
                }
                return new HasilBaris.Sah(kejadian);
            } catch (GalatTetapException e) {
                // Galat tetap tidak layak dicoba ulang: gagal cepat.
                jejak.add(new CatatanPercobaan(baris.nomor(), percobaan,
                        "gagal tetap, tanpa percobaan ulang: " + e.getMessage()));
                return new HasilBaris.GagalTetap(baris.nomor(), percobaan,
                        e.getMessage(), akar(e));
            } catch (GalatSementaraException e) {
                if (percobaan == kebijakan.percobaanMaks()) {
                    jejak.add(new CatatanPercobaan(baris.nomor(), percobaan,
                            "percobaan habis: " + e.getMessage()));
                    return new HasilBaris.GagalTetap(baris.nomor(), percobaan,
                            "percobaan ulang habis: " + e.getMessage(), akar(e));
                }
                long tunggu = kebijakan.unitTunggu(percobaan);
                totalTunggu.add(tunggu);
                jejak.add(new CatatanPercobaan(baris.nomor(), percobaan,
                        "gagal sementara: " + e.getMessage() + "; tunggu " + tunggu + " unit"));
            }
        }
        throw new IllegalStateException("tidak tercapai: loop selalu mengembalikan hasil");
    }

    /** Transformasi satu event; layanan simulasi melempar galat sesuai penanda. */
    private static KejadianTerproses prosesDenganLayanan(EventPenjualan event, int percobaan) {
        String penanda = event.penanda();
        if ("transien".equals(penanda) && percobaan == 1) {
            throw new GalatSementaraException("layanan tarif sedang sibuk");
        }
        if ("timeout".equals(penanda)) {
            throw new GalatSementaraException("layanan tarif tidak merespons",
                    new ConnectException("waktu tunggu koneksi habis"));
        }
        if ("rusak".equals(penanda)) {
            throw new GalatTetapException("sambungan gudang data terputus",
                    new ConnectException("koneksi ditolak oleh gudang data"));
        }
        return 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 utama dari ringkasan dan hasil per baris. */
    public static List<String> tulis(RingkasanAgregasi ringkasan, List<HasilBaris> hasil) {
        int sah = 0;
        int ditolak = 0;
        int gagal = 0;
        for (HasilBaris h : hasil) {
            switch (h) {
                case HasilBaris.Sah _ -> sah++;
                case HasilBaris.TidakSah _ -> ditolak++;
                case HasilBaris.GagalTetap _ -> gagal++;
            }
        }
        List<String> laporan = new ArrayList<>();
        laporan.add("=== Laporan pipeline penjualan ===");
        laporan.add("baris dibaca     : " + hasil.size());
        laporan.add("baris sah        : " + sah);
        laporan.add("ditolak validasi : " + ditolak);
        laporan.add("gagal tetap      : " + gagal);
        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)));
        return laporan;
    }

    /** Laporan galat terurut nomor baris: penolakan validasi dan kegagalan tetap. */
    public static List<String> tulisGalat(List<HasilBaris> hasil) {
        List<String> galat = new ArrayList<>();
        for (HasilBaris h : hasil) {
            switch (h) {
                case HasilBaris.Sah _ -> { }
                case HasilBaris.TidakSah(int nomor, String alasan) ->
                        galat.add(String.format(Locale.US,
                                "baris %2d  DITOLAK VALIDASI : %s", nomor, alasan));
                case HasilBaris.GagalTetap(int nomor, int percobaan, String pesan,
                        String penyebabAkar) ->
                        galat.add(String.format(Locale.US,
                                "baris %2d  GAGAL TETAP      : %s (percobaan %d; akar: %s)",
                                nomor, pesan, percobaan, penyebabAkar));
            }
        }
        return galat;
    }

    /** Menjalankan pipeline dengan percobaan ulang secara paralel pada eksekutor. */
    public static HasilPipeline jalankanDenganRetry(List<String> isiBerkas,
            ExecutorService eksekutor, KebijakanRetry kebijakan) {
        List<BarisData> baris = baca(isiBerkas);
        Map<Integer, HasilBaris> hasilPerNomor = new ConcurrentHashMap<>();
        // Jejak percobaan ditulis banyak thread sekaligus tanpa penguncian manual.
        ConcurrentLinkedQueue<CatatanPercobaan> jejak = new ConcurrentLinkedQueue<>();
        LongAdder totalTunggu = new LongAdder();
        List<CompletableFuture<Void>> tugas = new ArrayList<>(baris.size());
        for (BarisData b : baris) {
            tugas.add(CompletableFuture.runAsync(() -> hasilPerNomor.put(b.nomor(),
                    prosesDenganRetry(b, kebijakan, jejak, totalTunggu)), eksekutor));
        }
        CompletableFuture.allOf(tugas.toArray(new CompletableFuture<?>[0])).join();
        List<HasilBaris> terurut = new ArrayList<>(new TreeMap<>(hasilPerNomor).values());
        // Urutan kedatangan jejak tidak menentu: urutkan menurut baris lalu percobaan.
        List<CatatanPercobaan> jejakTerurut = jejak.stream()
                .sorted(Comparator.comparingInt(CatatanPercobaan::nomorBaris)
                        .thenComparingInt(CatatanPercobaan::percobaanKe))
                .toList();
        return rakitHasil(baris.size(), terurut, jejakTerurut, totalTunggu.sum());
    }

    /** Merakit hasil per baris menjadi statistik, laporan, dan laporan galat. */
    private static HasilPipeline rakitHasil(int jumlahBaris, List<HasilBaris> hasil,
            List<CatatanPercobaan> jejak, long unitTunggu) {
        List<KejadianTerproses> kejadian = new ArrayList<>(hasil.size());
        int ditolak = 0;
        int gagal = 0;
        for (HasilBaris h : hasil) {
            switch (h) {
                case HasilBaris.Sah(var k) -> kejadian.add(k);
                case HasilBaris.TidakSah _ -> ditolak++;
                case HasilBaris.GagalTetap _ -> gagal++;
            }
        }
        RingkasanAgregasi ringkasan = agregasi(kejadian);
        return new HasilPipeline(jumlahBaris, kejadian.size(), ditolak, gagal, unitTunggu,
                ringkasan.nilaiBerjasaPerKota().size(), ringkasan.jumlahPerKategori().size(),
                tulis(ringkasan, hasil), tulisGalat(hasil), jejak);
    }

    /** Merunut penyebab paling dalam dari rangkaian exception berantai. */
    private static String akar(Throwable galat) {
        Throwable kini = galat;
        while (kini.getCause() != null) {
            kini = kini.getCause();
        }
        return kini.getClass().getSimpleName() + ": " + kini.getMessage();
    }
}