StorageLayer.java

import java.sql.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class StorageLayer extends Thread {

    private static final String      DB_URL = "jdbc:sqlite:trades.db";
    private static final TradeRecord POISON = new TradeRecord(-1, -1, -1, 0, 0);

    private final BlockingQueue<TradeRecord> tradeQueue = new LinkedBlockingQueue<>();
    private boolean driverAvailable = false;

    public StorageLayer() {
        super("StorageLayer");
        setDaemon(true);
        loadDriver();
        if (driverAvailable) initDb();
    }

    public void record(TradeRecord trade) {
        if (driverAvailable) tradeQueue.offer(trade);
    }

    public void shutdown() {
        tradeQueue.offer(POISON);
    }

    @Override
    public void run() {
        if (!driverAvailable) {
            System.err.println("[Storage] Driver unavailable — storage disabled.");
            return;
        }

        try (Connection conn = DriverManager.getConnection(DB_URL)) {
            conn.setAutoCommit(false);

            String sql = "INSERT INTO trades(trade_id,buy_order_id,sell_order_id,price,quantity,executed_at) "
                       + "VALUES(?,?,?,?,?,?)";

            try (PreparedStatement stmt = conn.prepareStatement(sql)) {
                int batchCount = 0;

                while (true) {
                    TradeRecord t = tradeQueue.take();
                    if (t == POISON) break;

                    stmt.setInt   (1, t.tradeId);
                    stmt.setInt   (2, t.buyOrderId);
                    stmt.setInt   (3, t.sellOrderId);
                    stmt.setDouble(4, t.price);
                    stmt.setInt   (5, t.quantity);
                    stmt.setLong  (6, t.executedAt);
                    stmt.addBatch();
                    batchCount++;

                    if (batchCount % 500 == 0) {
                        stmt.executeBatch();
                        conn.commit();
                    }
                }

                stmt.executeBatch();
                conn.commit();
                System.out.println("[Storage] All trades flushed to trades.db");
            }

        } catch (SQLException | InterruptedException e) {
            System.err.println("[Storage] Error: " + e.getMessage());
        }
    }

    private void loadDriver() {
        try {
            Class.forName("org.sqlite.JDBC");
            driverAvailable = true;
        } catch (ClassNotFoundException e) {
            System.err.println("[Storage] sqlite-jdbc jar not found — trades will NOT be persisted.");
        }
    }

    private void initDb() {
        try (Connection conn = DriverManager.getConnection(DB_URL);
             Statement  stmt = conn.createStatement()) {
            stmt.execute(
                "CREATE TABLE IF NOT EXISTS trades (" +
                "  trade_id      INTEGER PRIMARY KEY," +
                "  buy_order_id  INTEGER NOT NULL," +
                "  sell_order_id INTEGER NOT NULL," +
                "  price         REAL    NOT NULL," +
                "  quantity      INTEGER NOT NULL," +
                "  executed_at   INTEGER NOT NULL" +
                ")"
            );
            System.out.println("[Storage] Database ready.");
        } catch (SQLException e) {
            System.err.println("[Storage] DB init error: " + e.getMessage());
        }
    }
}