jja 0.9.1

//
// jja: swiss army knife for chess file formats
// src/pgnbook.rs: Portable Game Notation parsing and import
//
// Copyright (c) 2023 Ali Polatel <alip@chesswob.org>
//
// SPDX-License-Identifier: GPL-3.0-or-later

use std::{
    collections::{HashMap, HashSet},
    fs::File,
    io::{Cursor, Read, Write},
    mem,
    path::Path,
    sync::{
        atomic::{AtomicBool, Ordering},
        Arc,
    },
};

use anyhow::{bail, Context};
use console::style;
use indicatif::ProgressBar;
use pgn_reader::{BufferedReader, RawHeader, SanPlus, Skip, Visitor};
use rayon::{
    iter::{IntoParallelRefIterator, ParallelIterator},
    ThreadPoolBuilder,
};
use shakmaty::{
    fen::{Epd, Fen},
    CastlingMode, Chess, Color, EnPassantMode, Outcome, ParseOutcomeError, Position, PositionError,
};

use crate::{
    hash::{zobrist_hash, ZobristHashSet, ZobristHasherBuilder},
    pgnfilt::{game_passes_filter, Field, FilterComponent},
    polyglot::from_move,
    tr,
};

/// A structure representing a game entry with move and game statistics.
#[derive(Copy, Clone, Default, Debug, Eq, PartialEq)]
pub struct GameEntry {
    /// The move encoded as a 16-bit unsigned integer.
    pub mov: u16,
    /// The total number of games played with this move.
    pub ngames: u64,
    /// The number of games won with this move.
    pub nwon: u64,
    /// The number of games lost with this move.
    pub nlost: u64,
}

/// A structure representing a game database, containing a path to the database and a book object.
pub struct GameBase {
    /// The path to the game database.
    pub path: Box<Path>,
    /// The opening book object containing game data, wrapped in an Arc for thread safety.
    pub book: Arc<rocksdb::DBWithThreadMode<rocksdb::MultiThreaded>>,
}

impl Drop for GameBase {
    fn drop(&mut self) {
        if let Err(e) = std::fs::remove_dir_all(&*self.path) {
            eprintln!("{}", tr!("Error removing the database directory: {}", e));
        }
    }
}

impl GameEntry {
    /// The constant size of the serialized representation of a `GameEntry`.
    pub const SERIALIZED_SIZE: usize = mem::size_of::<u16>() + 3 * mem::size_of::<u64>();

    /// Calculates the weight of a move, considering both won and drawn games.
    pub fn weight(&self, wdl_factor: &(f64, f64, f64)) -> u64 {
        let draws: u64 = self
            .ngames
            .saturating_sub(self.nwon.saturating_add(self.nlost));
        let w = (self.nwon as f64 * wdl_factor.0) as u64;
        let d = (draws as f64 * wdl_factor.1) as u64;
        let l = (self.nlost as f64 * wdl_factor.2) as u64;
        w.saturating_add(d).saturating_add(l)
    }

    /// Serializes a `GameEntry` instance into a binary format using a writer.
    pub fn serialize<W: Write>(&self, writer: &mut W) -> std::io::Result<()> {
        writer.write_all(&self.mov.to_le_bytes())?;
        writer.write_all(&self.ngames.to_le_bytes())?;
        writer.write_all(&self.nwon.to_le_bytes())?;
        writer.write_all(&self.nlost.to_le_bytes())?;
        Ok(())
    }

    /// Deserializes a `GameEntry` instance from a binary format using a reader.
    pub fn deserialize<R: Read>(reader: &mut R) -> std::io::Result<Self> {
        let mut mov_bytes = [0; mem::size_of::<u16>()];
        let mut ngames_bytes = [0; mem::size_of::<u64>()];
        let mut nwon_bytes = [0; mem::size_of::<u64>()];
        let mut nlost_bytes = [0; mem::size_of::<u64>()];

        reader.read_exact(&mut mov_bytes)?;
        reader.read_exact(&mut ngames_bytes)?;
        reader.read_exact(&mut nwon_bytes)?;
        reader.read_exact(&mut nlost_bytes)?;

        Ok(Self {
            mov: u16::from_le_bytes(mov_bytes),
            ngames: u64::from_le_bytes(ngames_bytes),
            nwon: u64::from_le_bytes(nwon_bytes),
            nlost: u64::from_le_bytes(nlost_bytes),
        })
    }
}

// Keep this struct small & packed, we send it between threads.
struct GameData {
    moves: Vec<(u64, u16, bool)>, // (Zobrist Hash, Polyglot 16-bit move repr, Color)
    result: Outcome,              // Game outcome
}

/// A structure representing a chess game, including its data, headers, and position.
pub struct Game {
    data: GameData,
    valid: bool,
    position: Chess,
    headers: HashMap<Field, String>,
}

/// Number of candidates moves on average.
/// Used by the vectors in merge operators.
pub const MERGE_INLINE_CAPACITY: usize = 16;

fn game_entry_merge(
    _key: &[u8],
    existing_val: Option<&[u8]>,
    operands: &rocksdb::MergeOperands,
) -> Option<Vec<u8>> {
    let mut merged_entries = Vec::with_capacity(MERGE_INLINE_CAPACITY);

    if let Some(existing_entry_bytes) = existing_val {
        let existing_entry = GameEntry::deserialize(&mut Cursor::new(existing_entry_bytes)).ok()?;
        merged_entries.push(existing_entry);
    }

    for operand in operands {
        let new_entry = GameEntry::deserialize(&mut Cursor::new(operand)).ok()?;

        // If the value is found then Result::Ok is returned, containing the index of the
        // matching element. If there are multiple matches, then any one of the matches
        // could be returned. The index is chosen deterministically, but is subject to
        // change in future versions of Rust. If the value is not found then Result::Err is
        // returned, containing the index where a matching element could be inserted while
        // maintaining sorted order.
        match merged_entries.binary_search_by(|probe| probe.mov.cmp(&new_entry.mov)) {
            Ok(index) => {
                merged_entries[index].ngames += new_entry.ngames;
                merged_entries[index].nwon += new_entry.nwon;
                merged_entries[index].nlost += new_entry.nlost;
            }
            Err(index) => {
                merged_entries.insert(index, new_entry);
            }
        }
    }

    let mut merged_entry_bytes = Vec::with_capacity(merged_entries.len());
    for entry in &merged_entries {
        entry.serialize(&mut merged_entry_bytes).unwrap();
    }
    Some(merged_entry_bytes)
}

fn game_entry_partial_merge(
    _key: &[u8],
    left: Option<&[u8]>,
    right: &rocksdb::MergeOperands,
) -> Option<Vec<u8>> {
    let mut merged_entries = Vec::with_capacity(MERGE_INLINE_CAPACITY);

    if let Some(left_entry_bytes) = left {
        let left_entry = GameEntry::deserialize(&mut Cursor::new(left_entry_bytes)).ok()?;
        merged_entries.push(left_entry);
    }

    for operand in right {
        let right_entry = GameEntry::deserialize(&mut Cursor::new(operand)).ok()?;

        // See note in game_entry_merge about binary_search_by()
        match merged_entries.binary_search_by(|probe| probe.mov.cmp(&right_entry.mov)) {
            Ok(index) => {
                merged_entries[index].ngames += right_entry.ngames;
                merged_entries[index].nwon += right_entry.nwon;
                merged_entries[index].nlost += right_entry.nlost;
            }
            Err(index) => {
                merged_entries.insert(index, right_entry);
            }
        }
    }

    let mut merged_entry_bytes = Vec::with_capacity(merged_entries.len());
    for entry in &merged_entries {
        entry.serialize(&mut merged_entry_bytes).unwrap();
    }
    Some(merged_entry_bytes)
}

/// Deserialize game entries from binary data into a `Vec` of `GameEntry` objects.
pub fn deserialize_game_entries(data: &[u8]) -> Vec<GameEntry> {
    let mut cursor = Cursor::new(data);
    let mut game_entries = Vec::with_capacity(MERGE_INLINE_CAPACITY);

    while let Ok(entry) = GameEntry::deserialize(&mut cursor) {
        game_entries.push(entry);
    }

    game_entries
}

/// A structure for creating an opening book, containing options and data for the book-making
/// process.
pub struct OpeningBookMaker {
    verbose: u8,
    games: usize,
    nfilt: usize,
    game: Game,
    max_ply: u64,
    min_pieces: usize,
    filter_side: Option<Color>,
    hashcode: Option<ZobristHashSet>,
    parsed_filter_expr: Option<Vec<FilterComponent>>,
    progress_bar: Option<ProgressBar>,
}

impl OpeningBookMaker {
    fn new(
        verbose: u8,
        max_ply: u64,
        min_pieces: usize,
        filter_side: Option<Color>,
        hashcode: bool,
        parsed_filter_expr: Option<Vec<FilterComponent>>,
        progress_bar: Option<ProgressBar>,
    ) -> OpeningBookMaker {
        OpeningBookMaker {
            verbose,
            games: 0,
            nfilt: 0,
            game: Game {
                data: GameData {
                    moves: Vec::new(),
                    result: Outcome::Draw,
                },
                valid: true,
                position: Chess::default(),
                headers: HashMap::with_capacity(18),
            },
            hashcode: if hashcode {
                Some(HashSet::with_hasher(ZobristHasherBuilder))
            } else {
                None
            },
            max_ply,
            min_pieces,
            filter_side,
            parsed_filter_expr,
            progress_bar,
        }
    }

    fn report_invalid_hashcode_error(&mut self, value: &[u8]) {
        // TODO: i18n!
        let val = String::from_utf8_lossy(value);
        let msg = format!(
            "Skipping invalid HashCode header `{}' in game {}",
            val, self.games
        );
        if let Some(progress_bar) = &self.progress_bar {
            progress_bar.println(&msg);
        } else {
            eprintln!("{}", msg);
        }
        self.game.valid = false;
    }
}

impl Visitor for OpeningBookMaker {
    type Result = Game;

    fn begin_game(&mut self) {
        self.games += 1;
    }

    fn header(&mut self, key: &[u8], value: RawHeader<'_>) {
        // Collect headers in a HashMap for filtering in end_headers.
        if self.parsed_filter_expr.is_some() {
            if let Some(field) = Field::from_utf8(key) {
                self.game
                    .headers
                    .insert(field, value.decode_utf8_lossy().to_string());
            }
        }

        // Support games from a non-standard starting position.
        if key == b"FEN" {
            let fen = match Fen::from_ascii(value.as_bytes()) {
                Ok(fen) => fen,
                Err(err) => {
                    let msg = tr!(
                        "Skipping invalid FEN header in game {}: {} ({}).",
                        self.games,
                        err,
                        format!("{:?}", value)
                    );
                    if let Some(progress_bar) = &self.progress_bar {
                        progress_bar.println(&msg);
                    } else {
                        eprintln!("{}", msg);
                    }
                    self.game.valid = false;
                    return;
                }
            };

            self.game.position = match fen
                .into_position(CastlingMode::Chess960)
                .or_else(PositionError::ignore_invalid_ep_square)
                .or_else(PositionError::ignore_invalid_castling_rights)
            {
                Ok(pos) => pos,
                Err(err) => {
                    let msg = format!(
                        "Skipping illegal FEN header in game {}: {} ({:?}).",
                        self.games, err, value
                    );
                    if let Some(progress_bar) = &self.progress_bar {
                        progress_bar.println(&msg);
                    } else {
                        eprintln!("{}", msg);
                    }
                    self.game.valid = false;
                    return;
                }
            };
        } else if key == b"Result" {
            let value = value.as_bytes();
            self.game.data.result = match Outcome::from_ascii(value) {
                Ok(outcome) => outcome,
                Err(ParseOutcomeError::Unknown) => {
                    self.game.valid = false;
                    return;
                }
                Err(err) => {
                    let msg = format!(
                        "Skipping invalid Result header `{:?}' in game {}: {}.",
                        value, self.games, err
                    );
                    if let Some(progress_bar) = &self.progress_bar {
                        progress_bar.println(&msg);
                    } else {
                        eprintln!("{}", msg);
                    }
                    self.game.valid = false;
                    return;
                }
            };
        } else if self.hashcode.is_some() && key == b"HashCode" {
            let value = value.as_bytes();
            if value.len() != 8 {
                self.report_invalid_hashcode_error(value);
                return;
            }
            let mut hcode = 0u64;
            for byte in value {
                match (*byte as char).to_digit(16) {
                    Some(val) => {
                        let (new_hcode, overflowed) = hcode.overflowing_shl(4);
                        if overflowed {
                            self.report_invalid_hashcode_error(value);
                            return;
                        }
                        hcode = new_hcode | u64::from(val);
                    }
                    None => {
                        self.report_invalid_hashcode_error(value);
                        return;
                    }
                }
            }

            // SAFETY: is_some() check above ensures hashcode is not None.
            let zhset = unsafe { self.hashcode.as_mut().unwrap_unchecked() };
            if !zhset.insert(hcode) {
                // HashCode already in set, skip game.
                self.game.valid = false;
                if self.verbose > 0 {
                    let msg = format!(
                        "Skipping duplicate game {} with HashCode `{:08x}'",
                        self.games, hcode
                    );
                    if let Some(progress_bar) = &self.progress_bar {
                        progress_bar.println(&msg);
                    } else {
                        eprintln!("{}", msg);
                    }
                }
                return;
            }
        }
    }

    fn end_headers(&mut self) -> Skip {
        if let Some(parsed_filter_expr) = &self.parsed_filter_expr {
            if !game_passes_filter(parsed_filter_expr, &self.game.headers, self.verbose) {
                self.game.valid = false; // skip game.
                self.nfilt += 1;
            } else if self.verbose > 0 {
                let msg = format!("Success matching header filter for game {}.", self.games);
                if let Some(progress_bar) = &self.progress_bar {
                    progress_bar.println(&msg);
                } else {
                    eprintln!("{}", msg);
                }
            }
        }
        Skip(!self.game.valid)
    }

    fn begin_variation(&mut self) -> Skip {
        Skip(true) // stay in the mainline
    }

    fn san(&mut self, san_plus: SanPlus) {
        if self.game.valid && self.max_ply as usize > self.game.data.moves.len() {
            if self.min_pieces > 2 /* no need to calculate for 2 and below */ &&
                self.min_pieces > self.game.position.board().occupied().count()
            {
                self.game.valid = false;
                return;
            }
            let mov = match san_plus.san.to_move(&self.game.position) {
                Ok(mov) => mov,
                Err(err) => {
                    let epd = format!(
                        "{}",
                        Epd::from_position(self.game.position.clone(), EnPassantMode::PseudoLegal)
                    );
                    let msg = format!(
                        "illegal SAN move `{}' in position `{}': {}",
                        san_plus.san, epd, err
                    );
                    if let Some(progress_bar) = &self.progress_bar {
                        progress_bar.println(&msg);
                    } else {
                        eprintln!("{}", msg);
                    }
                    self.game.valid = false;
                    return;
                }
            };

            // Filter based on side to move.
            let wtm = self.game.position.turn().is_white();
            if let Some(filter_side) = self.filter_side {
                if (wtm && filter_side != Color::White) || (!wtm && filter_side != Color::Black) {
                    self.game.position.play_unchecked(&mov);
                    return;
                }
            }

            let hash = zobrist_hash(&self.game.position);
            self.game.position.play_unchecked(&mov);
            self.game.data.moves.push((hash, from_move(&mov), wtm));
        }
    }

    fn end_game(&mut self) -> Self::Result {
        // Let's spare some capacity as we'll send this to another thread.
        self.game.data.moves.shrink_to_fit();

        mem::replace(
            &mut self.game,
            Game {
                data: GameData {
                    result: Outcome::Draw,
                    moves: Vec::with_capacity(self.max_ply as usize),
                },
                valid: true,
                position: Chess::default(),
                headers: HashMap::with_capacity(18),
            },
        )
    }
}

/// Create an opening book from a set of PGN files.
///
/// # Arguments
/// * `pgn_files`: A list of PGN file paths.
/// * `db_path`: Path to the output Polyglot opening book.
/// * `verbose`: Level of verbosity
/// * `max_ply`: Maximum number of plies to include in the opening book.
/// * `min_pieces`: Minimum number of pieces on the board for the position
/// to be included in the book.
/// * `filter_side`: Given a colour, only includes only moves with that colour to move.
/// * `hashcode`: Deduplicate games using the `HashCode` PGN tag
/// * `threads`: Number of threads to use for parallel processing.
/// * `batch_size`: Write batch size in number of games.
/// * `compression_which`: Compression type to use for the rocksdb database.
/// * `compression_level`: The compression level to use for the rocksdb database.
/// * `max_open_files`: Maximum number of open files for the rocksdb database.
/// * `parsed_filter_expr`: Optional filter expression to filter headers.
///
/// # Returns
/// An `Result` containing a `GameBase` representing the opening book, or an error if there was a
/// problem creating the book.
#[allow(clippy::too_many_arguments)]
pub fn create_opening_book(
    pgn_files: &[String],
    db_path: &Path,
    verbose: u8,
    max_ply: u64,
    min_pieces: usize,
    filter_side: Option<Color>,
    hashcode: bool,
    threads: usize,
    batch_size: usize,
    compression_which: rocksdb::DBCompressionType,
    compression_level: i32,
    max_open_files: i32,
    parsed_filter_expr: Option<Vec<FilterComponent>>,
    progress_bar: Option<ProgressBar>,
) -> anyhow::Result<GameBase> {
    let mut db_options = rocksdb::Options::default();
    db_options.create_if_missing(true);
    db_options.set_error_if_exists(true);

    db_options.increase_parallelism(threads as i32);
    db_options.set_max_background_jobs((threads as i32 + 1) / 2);

    // Memtable, 64MB * threads
    db_options.set_write_buffer_size(0x4000000 * threads);

    // Prevents too many open files
    db_options.set_max_open_files(max_open_files.saturating_mul(threads as i32));

    // Prepare for bulk load.
    // 1. set max background jobs to at least 4
    // 2. disables autocompaction, should issue a manual compaction afterwards.
    // 3. Set memtable to Vector (does not support concurrent writes).
    db_options.prepare_for_bulk_load();

    db_options.set_memtable_factory(rocksdb::MemtableFactory::Vector);
    db_options.set_allow_concurrent_memtable_write(false);

    // Compression
    db_options.set_compression_type(compression_which);
    db_options.set_compression_per_level(&[
        compression_which,
        compression_which,
        compression_which,
        compression_which,
        compression_which,
    ]);
    db_options.set_compression_options(-14, compression_level, 0, 0);
    db_options.set_bottommost_compression_options(-14, compression_level, 0, 0, true);

    // Merge operators
    db_options.set_merge_operator(
        "game_entry_merge",
        game_entry_merge,
        game_entry_partial_merge,
    );

    /*
    if verbose > 1 {
        db_options.set_log_level(rocksdb::LogLevel::Debug);
    } else if verbose > 0 {
        db_options.set_log_level(rocksdb::LogLevel::Info);
    }
    */

    let opening_book = rocksdb::DB::open(&db_options, db_path)
        .with_context(|| tr!("Failed to open RocksDB database `{}'.", db_path.display()))?;
    opening_book
        .create_cf("opening_book", &db_options)
        .with_context(|| {
            tr!(
                "Failed to create column family in RocksDB database `{}'.",
                db_path.display()
            )
        })?;

    let opening_book = Arc::new(opening_book);
    let opening_base = GameBase {
        path: db_path.into(),
        book: Arc::clone(&opening_book),
    };

    /* Handle interruptions */
    let interrupted = Arc::new(AtomicBool::new(false));

    let interrupted_clone = Arc::clone(&interrupted);
    ctrlc::set_handler(move || {
        interrupted_clone.store(true, Ordering::SeqCst);
    })
    .context(tr!("Failed to set up interrupt handler."))?;

    // Distribute the thread count evenly between book and parser threads.
    let mut thread_count_ppgn = (threads + 1) / 2; // Add 1 to the numerator to ensure rounding up
    let mut thread_count_book = threads.saturating_sub(thread_count_ppgn);
    if thread_count_ppgn == 0 || thread_count_book == 0 {
        thread_count_ppgn = 1;
        thread_count_book = 1;
    };

    let mut games = 0; // Total number of games.
    let mut nfilt = 0; // Total number of games filtered out.
    let (send, recv) =
        crossbeam::channel::bounded::<GameData>(batch_size.saturating_mul(thread_count_book));

    let mut book_threads = Vec::new();
    for i in 0..thread_count_book {
        let recv = recv.clone();
        let opening_book = Arc::clone(&opening_book);
        let interrupted_clone = Arc::clone(&interrupted);
        let book_thread = std::thread::Builder::new()
            .name(format!("jja-book-{}", i))
            .spawn(move || {
                let mut batch = rocksdb::WriteBatch::default();
                let mut batch_counter = 0;
                let opening_book_cf = opening_book.cf_handle("opening_book").unwrap();

                for game_data in recv {
                    if interrupted_clone.load(Ordering::SeqCst) {
                        break;
                    }

                    for (hash, mov, wtm) in game_data.moves {
                        let new_entry = GameEntry {
                            mov,
                            ngames: 1,
                            nwon: if (game_data.result
                                == Outcome::Decisive {
                                    winner: Color::White,
                                }
                                && wtm)
                                || (game_data.result
                                    == Outcome::Decisive {
                                        winner: Color::Black,
                                    }
                                    && !wtm)
                            {
                                1
                            } else {
                                0
                            },
                            nlost: if (game_data.result
                                == Outcome::Decisive {
                                    winner: Color::White,
                                }
                                && !wtm)
                                || (game_data.result
                                    == Outcome::Decisive {
                                        winner: Color::Black,
                                    }
                                    && wtm)
                            {
                                1
                            } else {
                                0
                            },
                        };

                        let key = hash.to_le_bytes();
                        let mut new_entry_bytes = Vec::new();
                        new_entry.serialize(&mut new_entry_bytes).unwrap();

                        batch.merge_cf(&opening_book_cf, key, &new_entry_bytes);
                        batch_counter += 1;

                        if batch_counter >= batch_size {
                            opening_book
                                .write(batch)
                                .expect("Failed to write batch merge operation");
                            batch = rocksdb::WriteBatch::default();
                            batch_counter = 0;
                        }
                    }
                }

                if batch_counter > 0 {
                    opening_book
                        .write(batch)
                        .expect("Failed to write batch merge operation");
                }
            })
            .context(tr!("Failed to spawn book thread `{}'.", i))?;
        book_threads.push(book_thread);
    }

    // Create a ThreadPool with the desired number of threads
    let parser_pool = ThreadPoolBuilder::new()
        .num_threads(thread_count_ppgn)
        .thread_name(|i| format!("jja-parser-{}", i))
        .build()
        .context(tr!("Failed to build PGN parser thread pool."))?;

    let pgn_files_par_iter = parser_pool.install(|| {
        pgn_files
            .par_iter()
            .map(|arg| -> Result<(usize, usize), String> {
                let mut opening_book_maker = OpeningBookMaker::new(
                    verbose,
                    max_ply,
                    min_pieces,
                    filter_side,
                    hashcode,
                    parsed_filter_expr.clone(),
                    progress_bar.clone(),
                );

                if let Some(ref progress_bar) = progress_bar {
                    progress_bar.println(tr!("Opening input PGN file `{}'...", style(arg).bold().green()));
                }
                let file = match File::open(arg) {
                    Ok(file) => file,
                    Err(err) => {
                        interrupted.store(true, Ordering::SeqCst);
                        return Err(tr!("Failed to open PGN file `{}': {}", arg, err));
                    }
                };

                let ext = std::path::Path::new(arg).extension();
                let uncompressed: Box<dyn Read + Send> = if ext.map_or(false, |ext| ext.eq_ignore_ascii_case("zst")) {
                    Box::new(match zstd::Decoder::new(file) {
                        Ok(decoder) => decoder,
                        Err(err) => {
                            interrupted.store(true, Ordering::SeqCst);
                            return Err(tr!("Failed to open PGN file `{}': {}", arg, err));
                        }
                    })
                } else if ext.map_or(false, |ext| ext.eq_ignore_ascii_case("zst")) {
                    Box::new(bzip2::read::MultiBzDecoder::new(file))
                } else if ext.map_or(false, |ext| ext.eq_ignore_ascii_case("zst")) {
                    Box::new(xz2::read::XzDecoder::new(file))
                } else if ext.map_or(false, |ext| ext.eq_ignore_ascii_case("zst")) {
                    Box::new(flate2::read::GzDecoder::new(file))
                } else if ext.map_or(false, |ext| ext.eq_ignore_ascii_case("zst")) {
                    Box::new(match lz4::Decoder::new(file) {
                        Ok(decoder) => decoder,
                        Err(err) => {
                            interrupted.store(true, Ordering::SeqCst);
                            return Err(tr!("Failed to open PGN file `{}': {}", arg, err));
                        }
                    })
                } else {
                    Box::new(file)
                };

                for game in BufferedReader::new(uncompressed).into_iter(&mut opening_book_maker) {
                    if interrupted.load(Ordering::SeqCst) {
                        break;
                    }
                    match game {
                        Ok(game) => {
                            if !game.data.moves.is_empty() {
                                send.send(game.data).unwrap();
                            }
                            if let Some(ref progress_bar) = progress_bar {
                                progress_bar.inc(1);
                            }
                        }
                        Err(err) => {
                            let msg = tr!(
                                "Stopping to parse PGN input file `{}' at game {} due to error: {}.",
                                style(arg).bold().green(),
                                style(opening_book_maker.games).bold().cyan(),
                                style(err).bold().red()
                            );
                            if let Some(progress_bar) = &progress_bar {
                                progress_bar.println(&msg);
                            } else {
                                eprintln!("{}", msg);
                            }
                            break;
                        }
                    };
                }
                Ok((opening_book_maker.games, opening_book_maker.nfilt))
            })
            .collect::<Result<Vec<(usize, usize)>, String>>()
    });

    // Update the total games and nfilt counters
    match pgn_files_par_iter {
        Ok(results) => {
            for (local_games, local_nfilt) in results {
                games += local_games;
                nfilt += local_nfilt;
            }
        }
        Err(err) => {
            if let Some(ref progress_bar) = progress_bar {
                progress_bar.println(tr!("Failed to parse PGN: {}.", style(err).bold().red()));
            }
            /* interrupted is already true here, nothing else to do. */
        }
    }

    drop(send); // Close the sender so receivers can exit gracefully.
    for book_thread in book_threads {
        book_thread.join().expect("join book thread");
    }

    if interrupted.load(Ordering::SeqCst) {
        if let Some(progress_bar) = progress_bar {
            progress_bar.finish_and_clear();
        }
        eprintln!("{}", tr!("Failed to read all games, process interrupted."));
        eprintln!(
            "{}",
            tr!("Cleaning up the temporary database. This may take a while.")
        );
        bail!("{}", tr!("Operation was interrupted."));
    }

    if let Some(progress_bar) = progress_bar {
        progress_bar.println(tr!(
            "Success reading {} games, with {} filtered out, in {} input PGN files.",
            style(games).bold().cyan(),
            style(nfilt).red(),
            style(pgn_files.len()).bold().cyan()
        ));
        progress_bar.println(tr!(
            "Compacting the temporary rocksdb database. This may take a while."
        ));
    }

    // Note about prepare_for_bulk_load();
    // It's recommended to manually call CompactRange(NULL, NULL) before reading from the database,
    // because otherwise the read can be very slow.
    opening_book.compact_range(None::<&[u8]>, None::<&[u8]>);

    Ok(opening_base)
}

#[cfg(test)]
mod tests {
    use quick_csv::Csv;
    use shakmaty::{uci::Uci, Square};

    fn get_jja() -> std::process::Command {
        test_bin::get_test_bin("jja")
    }

    #[test]
    fn pgn_make_test_001() {
        let root = std::path::Path::new(env!("CARGO_MANIFEST_DIR"));
        let temp = tempfile::tempdir().expect("tempdir");

        let src = root.join("samples").join("sample-1.pgn");
        let dst = temp.path().join("book-1.bin");

        let src = src.to_str().expect("src");
        let dst = dst.to_str().expect("dst");

        let r = get_jja()
            .args(["make", src, "--min-games=1", "-o", dst])
            .status()
            .expect("execute jja-make");
        assert!(r.success());

        let o = get_jja()
            .args(["find", dst])
            .output()
            .expect("execute jja-find");
        assert!(o.status.success());

        let reader = Csv::from_reader(o.stdout.as_slice())
            .delimiter(b',')
            .has_header(true);

        for record_result in reader {
            let record = record_result.expect("csv record");
            let mut columns = record.columns().expect("cannot convert csv record to utf8");

            let uci =
                Uci::from_ascii(columns.nth(1).expect("uci column").as_bytes()).expect("uci move");
            let weight = str::parse::<u16>(columns.next().expect("weight column")).expect("weight");

            assert_eq!(
                uci,
                Uci::Normal {
                    from: Square::E2,
                    to: Square::E4,
                    promotion: None
                }
            );
            assert_eq!(weight, 65520);

            break;
        }
    }
}