<commit_before>package com.github.davidmoten.fsm.persistence;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;

import com.github.davidmoten.fsm.runtime.Clock;
import com.github.davidmoten.fsm.runtime.EntityBehaviour;
import com.github.davidmoten.fsm.runtime.EntityState;
import com.github.davidmoten.fsm.runtime.EntityStateMachine;
import com.github.davidmoten.fsm.runtime.Event;
import com.github.davidmoten.fsm.runtime.Signal;

public final class PersistenceH2 implements Persistence {

    private final File directory;
    private final ScheduledExecutorService executor;
    private final Clock clock;
    private final Serializer entitySerializer;
    private final Serializer eventSerializer;
    private final Queue<NumberedSignal<?, ?>> queue = new LinkedList<>();
    private final AtomicInteger wip = new AtomicInteger();
    private final Function<Class<?>, EntityBehaviour<?, String>> behaviourFactory;

    public PersistenceH2(File directory, ScheduledExecutorService executor, Clock clock,
            Serializer entitySerializer, Serializer eventSerializer,
            Function<Class<?>, EntityBehaviour<?, String>> behaviourFactory) {
        this.directory = directory;
        this.executor = executor;
        this.clock = clock;
        this.entitySerializer = entitySerializer;
        this.eventSerializer = eventSerializer;
        this.behaviourFactory = behaviourFactory;
    }

    public void create() {
        directory.mkdirs();
        try (Connection con = createConnection()) {
            con.setAutoCommit(true);
            String sql = new String(
                    readAll(PersistenceH2.class.getResourceAsStream("/create-h2.sql")),
                    StandardCharsets.UTF_8);
            String[] commands = sql.split(";");
            for (String command : commands) {
                con.prepareStatement(command).execute();
            }
            con.commit();
        } catch (SQLException e) {
            throw new SQLRuntimeException(e);
        }
    }

    private static final class NumberedDelayedSignal<T, Id> {
        final NumberedSignal<T, Id> signal;
        final long time;

        NumberedDelayedSignal(NumberedSignal<T, Id> signal, long time) {
            this.signal = signal;
            this.time = time;
        }
    }

    @SuppressWarnings("unchecked")
    public void initialize() {
        List<NumberedDelayedSignal<?, ?>> list = new ArrayList<NumberedDelayedSignal<?, ?>>();
        try (Connection con = createConnection()) {
            try (PreparedStatement ps = con.prepareStatement(
                    "select seq_num, cls, id, event_bytes, time from delayed_signal_queue order by seq_num")) {
                try (ResultSet rs = ps.executeQuery()) {
                    while (rs.next()) {
                        long number = rs.getLong("seq_num");
                        String className = rs.getString("cls");
                        String id = rs.getString("id");
                        byte[] eventBytes = readAll(rs.getBlob("event_bytes").getBinaryStream());
                        Object event = eventSerializer.deserialize(eventBytes);
                        Class<?> cls = Class.forName(className);
                        long time = rs.getTimestamp("times").getTime();
                        Signal<Object, String> signal = Signal.create((Class<Object>) cls, id,
                                (Event<Object>) event);
                        list.add(new NumberedDelayedSignal<Object, String>(
                                new NumberedSignal<Object, String>(signal, number), time));
                    }
                } catch (ClassNotFoundException e) {
                    throw new RuntimeException(e);
                }
            }
        } catch (SQLException e) {
            throw new SQLRuntimeException(e);
        }
        for (NumberedDelayedSignal<?, ?> sig : list) {
            long now = clock.now();
            long delayMs = Math.max(0, sig.time - now);
            executor.schedule(() -> offer((NumberedSignal<?, String>) sig.signal), delayMs,
                    TimeUnit.MILLISECONDS);
        }
    }

    private static byte[] readAll(InputStream is) {
        ByteArrayOutputStream bytes = new ByteArrayOutputStream();
        byte[] buffer = new byte[8192];
        int count;
        try {
            while ((count = is.read(buffer)) != -1) {
                bytes.write(buffer, 0, count);
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return bytes.toByteArray();
    }

    private static final class EntityAndState<T> {
        final T entity;
        final EntityState<T> state;

        EntityAndState(T entity, EntityState<T> state) {
            this.entity = entity;
            this.state = state;
        }

        static <T> EntityAndState<T> create(T entity, EntityState<T> state) {
            return new EntityAndState<T>(entity, state);
        }
    }

    @SuppressWarnings("unchecked")
    private void process(NumberedSignal<?, String> signal) {
        List<NumberedSignal<?, ?>> numberedSignalsToOther;
        try (Connection con = createConnection()) {
            con.setAutoCommit(false);

            // get behaviour
            EntityBehaviour<Object, String> behaviour = (EntityBehaviour<Object, String>) behaviourFactory
                    .apply(signal.signal.cls());

            // read entity
            Optional<EntityAndState<Object>> entity = readEntity(con,
                    (Class<Object>) signal.signal.cls(), signal.signal.id(),
                    (EntityBehaviour<Object, String>) behaviour);

            // initialize state machine
            final EntityStateMachine<?, String> esm;
            if (entity.isPresent()) {
                esm = behaviour.create(signal.signal.id());
            } else {
                esm = behaviour.create(signal.signal.id(), entity.get().entity, entity.get().state);
            }

            // apend signal to signal_store
            insertIntoSignalStore(esm, signal.signal.event(), eventSerializer(), con);

            // push signal through state machine
            EntityStateMachine<?, String> esm2 = esm.signal((Event<Object>) signal.signal.event());

            // add signals to others to signal_queue
            List<Signal<?, ?>> signalsToOther = esm2.signalsToOther();
            numberedSignalsToOther = insertSignalsToOther(eventSerializer(), con, signalsToOther);

            // add delayed signals to other to delayed_signal_queue
            insertDelayedSignalsToOther(eventSerializer(), con, signalsToOther);

            // remove signal from signal_queue
            removeSignal(signal.number, con);

            // update/create the entity bytes and state to entity table
            saveEntity(con, esm2);

            // commit the transaction
            con.commit();
        } catch (SQLException e) {
            throw new SQLRuntimeException(e);
        }
        for (NumberedSignal<?, ?> signalToOther : numberedSignalsToOther) {
            offer((NumberedSignal<?, String>) signalToOther);
        }
        // TODO call executor with delayed signals
    }

    private void saveEntity(Connection con, EntityStateMachine<?, String> esm) throws SQLException {
        final boolean updated;
        try (PreparedStatement ps = con
                .prepareStatement("update entity set bytes=?, state=? where cls=? and id=?")) {
            byte[] bytes = entitySerializer.serialize(esm.get().get());
            ps.setBlob(1, new ByteArrayInputStream(bytes));
            ps.setString(2, esm.state().toString());<commit_msg>persistence dev<commit_after>package com.github.davidmoten.fsm.persistence;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;

import com.github.davidmoten.fsm.runtime.Clock;
import com.github.davidmoten.fsm.runtime.EntityBehaviour;
import com.github.davidmoten.fsm.runtime.EntityState;
import com.github.davidmoten.fsm.runtime.EntityStateMachine;
import com.github.davidmoten.fsm.runtime.Event;
import com.github.davidmoten.fsm.runtime.Signal;

public final class PersistenceH2 implements Persistence {

    private final File directory;
    private final ScheduledExecutorService executor;
    private final Clock clock;
    private final Serializer entitySerializer;
    private final Serializer eventSerializer;
    private final Queue<NumberedSignal<?, ?>> queue = new LinkedList<>();
    private final AtomicInteger wip = new AtomicInteger();
    private final Function<Class<?>, EntityBehaviour<?, String>> behaviourFactory;

    public PersistenceH2(File directory, ScheduledExecutorService executor, Clock clock,
            Serializer entitySerializer, Serializer eventSerializer,
            Function<Class<?>, EntityBehaviour<?, String>> behaviourFactory) {
        this.directory = directory;
        this.executor = executor;
        this.clock = clock;
        this.entitySerializer = entitySerializer;
        this.eventSerializer = eventSerializer;
        this.behaviourFactory = behaviourFactory;
    }

    public void create() {
        directory.mkdirs();
        try (Connection con = createConnection()) {
            con.setAutoCommit(true);
            String sql = new String(
                    readAll(PersistenceH2.class.getResourceAsStream("/create-h2.sql")),
                    StandardCharsets.UTF_8);
            String[] commands = sql.split(";");
            for (String command : commands) {
                con.prepareStatement(command).execute();
            }
            con.commit();
        } catch (SQLException e) {
            throw new SQLRuntimeException(e);
        }
    }

    private static final class NumberedDelayedSignal<T, Id> {
        final NumberedSignal<T, Id> signal;
        final long time;

        NumberedDelayedSignal(NumberedSignal<T, Id> signal, long time) {
            this.signal = signal;
            this.time = time;
        }
    }

    @SuppressWarnings("unchecked")
    public void initialize() {
        List<NumberedDelayedSignal<?, ?>> list = new ArrayList<NumberedDelayedSignal<?, ?>>();
        try (Connection con = createConnection()) {
            try (PreparedStatement ps = con.prepareStatement(
                    "select seq_num, cls, id, event_bytes, time from delayed_signal_queue order by seq_num")) {
                try (ResultSet rs = ps.executeQuery()) {
                    while (rs.next()) {
                        long number = rs.getLong("seq_num");
                        String className = rs.getString("cls");
                        String id = rs.getString("id");
                        byte[] eventBytes = readAll(rs.getBlob("event_bytes").getBinaryStream());
                        Object event = eventSerializer.deserialize(eventBytes);
                        Class<?> cls = Class.forName(className);
                        long time = rs.getTimestamp("times").getTime();
                        Signal<Object, String> signal = Signal.create((Class<Object>) cls, id,
                                (Event<Object>) event);
                        list.add(new NumberedDelayedSignal<Object, String>(
                                new NumberedSignal<Object, String>(signal, number), time));
                    }
                } catch (ClassNotFoundException e) {
                    throw new RuntimeException(e);
                }
            }
        } catch (SQLException e) {
            throw new SQLRuntimeException(e);
        }
        for (NumberedDelayedSignal<?, ?> sig : list) {
            schedule(sig.signal, sig.time);
        }
    }

    @SuppressWarnings("unchecked")
    private void schedule(NumberedSignal<?, ?> sig, long time) {
        long now = clock.now();
        long delayMs = Math.max(0, time - now);
        executor.schedule(() -> offer((NumberedSignal<?, String>) sig), delayMs,
                TimeUnit.MILLISECONDS);
    }

    private static byte[] readAll(InputStream is) {
        ByteArrayOutputStream bytes = new ByteArrayOutputStream();
        byte[] buffer = new byte[8192];
        int count;
        try {
            while ((count = is.read(buffer)) != -1) {
                bytes.write(buffer, 0, count);
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        return bytes.toByteArray();
    }

    private static final class EntityAndState<T> {
        final T entity;
        final EntityState<T> state;

        EntityAndState(T entity, EntityState<T> state) {
            this.entity = entity;
            this.state = state;
        }

        static <T> EntityAndState<T> create(T entity, EntityState<T> state) {
            return new EntityAndState<T>(entity, state);
        }
    }

    @SuppressWarnings("unchecked")
    private void process(NumberedSignal<?, String> signal) {
        List<NumberedSignal<?, ?>> numberedSignalsToOther;
        try (Connection con = createConnection()) {
            con.setAutoCommit(false);

            // get behaviour
            EntityBehaviour<Object, String> behaviour = (EntityBehaviour<Object, String>) behaviourFactory
                    .apply(signal.signal.cls());

            // read entity
            Optional<EntityAndState<Object>> entity = readEntity(con,
                    (Class<Object>) signal.signal.cls(), signal.signal.id(),
                    (EntityBehaviour<Object, String>) behaviour);

            // initialize state machine
            final EntityStateMachine<?, String> esm;
            if (entity.isPresent()) {
                esm = behaviour.create(signal.signal.id());
            } else {
                esm = behaviour.create(signal.signal.id(), entity.get().entity, entity.get().state);
            }

            // apend signal to signal_store
            insertIntoSignalStore(esm, signal.signal.event(), eventSerializer(), con);

            // push signal through state machine
            EntityStateMachine<?, String> esm2 = esm.signal((Event<Object>) signal.signal.event());

            // add signals to others to signal_queue
            List<Signal<?, ?>> signalsToOther = esm2.signalsToOther();
            numberedSignalsToOther = insertSignalsToOther(eventSerializer(), con, signalsToOther);

            // add delayed signals to other to delayed_signal_queue
            insertDelayedSignalsToOther(eventSerializer(), con, signalsToOther);
            
            //TODO handle delayed signal to self

            // remove signal from signal_queue
            removeSignal(signal.number, con);

            // update/create the entity bytes and state to entity table
            saveEntity(con, esm2);

            // commit the transaction
            con.commit();
        } catch (SQLException e) {
            throw new SQLRuntimeException(e);
        }
        for (NumberedSignal<?, ?> signalToOther : numberedSignalsToOther) {
            offer((NumberedSignal<?, String>) signalToOther);
        }
        // TODO call executor with delayed signals
    }

    private void saveEntity(Connection con, EntityStateMachine<?, String> esm) throws SQLException {
        final boolean updated;
        try (PreparedStatement ps = con
                .prepareStatement("update entity set bytes=?, state=? where cls=? and id=?")) {
            byte[] bytes = entitySerializer.serialize(esm.get().get());
            ps.setBlob(1, new ByteArrayInputStream(bytes));
            ps.setString(2, esm.state().toString());
{'fraction_non_alphanumeric': 0.10505303301156001, 'fraction_numerical': 0.0020259802168990583, 'mean_word_length': 2.2933673469387754, 'pattern_counts': {'":': 0, '<': 100, '<?xml version=': 0, '>': 102, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1469512', 'n_tokens_mistral': 4329, 'n_tokens_neox': 4210, 'n_words': 1021}