<commit_before>package jet.bpm.engine.leveldb.index;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import jet.bpm.engine.event.ExpiredEvent;
import jet.bpm.engine.leveldb.LevelDb;
import jet.bpm.engine.leveldb.PersistentEvent;
import jet.bpm.engine.leveldb.Serializer;
import org.iq80.leveldb.DBIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ExpiredEventIndex {

    private static final Logger log = LoggerFactory.getLogger(ExpiredEventIndex.class);

    private final String currentMarker = UUID.randomUUID().toString();

    private final Serializer serializer;

    private final LevelDb levelDb;

    public ExpiredEventIndex(LevelDb levelDb, Serializer serializer) {
        this.levelDb = levelDb;
        this.serializer = serializer;
    }

    public void init() {
        levelDb.init();
    }

    public void close() {
        levelDb.close();
    }

    public void onAdd(PersistentEvent event) {
        Date expiredAt = event.getEvent().getExpiredAt();
        if (expiredAt == null) {
            return;
        }

        String processBusinessKey = event.getEvent().getProcessBusinessKey();
        String eventName = event.getEvent().getName();

        byte[] key = marshallKey(new IndexKey(event.getId(), expiredAt.getTime()));
        byte[] value = marshallValue(new IndexValue(processBusinessKey, eventName, expiredAt));
        levelDb.put(key, value);
    }

    public void onRemove(PersistentEvent event) {
        Date expiredAt = event.getEvent().getExpiredAt();
        if (expiredAt == null) {
            return;
        }

        byte[] key = marshallKey(new IndexKey(event.getId(), expiredAt.getTime()));
        levelDb.delete(key);
    }

    public List<ExpiredEvent> list(Date now, int maxEventsCount) {
        List<ExpiredEvent> result = new ArrayList<>();

        long nowTime = now.getTime();

        Map<IndexKey, IndexValue> processedItems = new HashMap<>();
        try (DBIterator it = levelDb.iterator();) {
            for (it.seekToFirst(); it.hasNext();) {
                Map.Entry<byte[], byte[]> entry = it.next();

                IndexKey k = unmarshallKey(entry.getKey());
                if (k.getExpiredAt() > nowTime) {
                    break;
                }

                if (result.size() >= maxEventsCount) {
                    break;
                }

                IndexValue v = unmarshallValue(entry.getValue());

                boolean isInProcess = currentMarker.equals(v.getProcessedMarker());

                if(!isInProcess) {
                    result.add(new ExpiredEvent(v.getProcessBusinessKey(), v.getEventName(), v.getExpiredAt()));
                    processedItems.put(k, v);
                } else {
                    log.warn("list ['{}', {}] -> found items in process", now, maxEventsCount);
                }
            }

            for (Map.Entry<IndexKey, IndexValue> item : processedItems.entrySet()) {
                IndexKey k = item.getKey();
                IndexValue v = item.getValue();

                v.setProcessedMarker(currentMarker);

                levelDb.put(marshallKey(k), marshallValue(v));
            }

            log.info("list ['{}', {}] -> done ({})", now, maxEventsCount, result.size());

            return result;
        } catch (Exception e) {
            log.error("list ['{}', {}] -> error", now, maxEventsCount, e);
            throw new RuntimeException("call 'list' error");
        }
    }

    private byte[] marshallKey(IndexKey key) {
        UUID id = key.getEventId();
        long expiredAt = key.getExpiredAt();

        long mostSigBits = id.getMostSignificantBits();
        long leastSigBits = id.getLeastSignificantBits();
        return ByteBuffer.allocate(8 + 8 + 8)
                .putLong(expiredAt)
                .putLong(mostSigBits)
                .putLong(leastSigBits)
                .array();
    }

    private IndexKey unmarshallKey(byte[] key) {
        ByteBuffer buffer = ByteBuffer.allocate(8 + 8 + 8).put(key);
        buffer.flip();
        long expiredAt = buffer.getLong();
        long mostSigBits = buffer.getLong();
        long leastSigBits = buffer.getLong();

        return new IndexKey(new UUID(mostSigBits, leastSigBits), expiredAt);
    }<commit_msg>fix inprocess items notification<commit_after>package jet.bpm.engine.leveldb.index;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import jet.bpm.engine.event.ExpiredEvent;
import jet.bpm.engine.leveldb.LevelDb;
import jet.bpm.engine.leveldb.PersistentEvent;
import jet.bpm.engine.leveldb.Serializer;
import org.iq80.leveldb.DBIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ExpiredEventIndex {

    private static final Logger log = LoggerFactory.getLogger(ExpiredEventIndex.class);

    private final String currentMarker = UUID.randomUUID().toString();

    private final Serializer serializer;

    private final LevelDb levelDb;

    public ExpiredEventIndex(LevelDb levelDb, Serializer serializer) {
        this.levelDb = levelDb;
        this.serializer = serializer;
    }

    public void init() {
        levelDb.init();
    }

    public void close() {
        levelDb.close();
    }

    public void onAdd(PersistentEvent event) {
        Date expiredAt = event.getEvent().getExpiredAt();
        if (expiredAt == null) {
            return;
        }

        String processBusinessKey = event.getEvent().getProcessBusinessKey();
        String eventName = event.getEvent().getName();

        byte[] key = marshallKey(new IndexKey(event.getId(), expiredAt.getTime()));
        byte[] value = marshallValue(new IndexValue(processBusinessKey, eventName, expiredAt));
        levelDb.put(key, value);
    }

    public void onRemove(PersistentEvent event) {
        Date expiredAt = event.getEvent().getExpiredAt();
        if (expiredAt == null) {
            return;
        }

        byte[] key = marshallKey(new IndexKey(event.getId(), expiredAt.getTime()));
        levelDb.delete(key);
    }

    public List<ExpiredEvent> list(Date now, int maxEventsCount) {
        List<ExpiredEvent> result = new ArrayList<>();

        long nowTime = now.getTime();

        int inProcessItemsCount = 0;
        Map<IndexKey, IndexValue> processedItems = new HashMap<>();
        try (DBIterator it = levelDb.iterator();) {
            for (it.seekToFirst(); it.hasNext();) {
                Map.Entry<byte[], byte[]> entry = it.next();

                IndexKey k = unmarshallKey(entry.getKey());
                if (k.getExpiredAt() > nowTime) {
                    break;
                }

                if (result.size() >= maxEventsCount) {
                    break;
                }

                IndexValue v = unmarshallValue(entry.getValue());

                boolean isInProcess = currentMarker.equals(v.getProcessedMarker());

                if(!isInProcess) {
                    result.add(new ExpiredEvent(v.getProcessBusinessKey(), v.getEventName(), v.getExpiredAt()));
                    processedItems.put(k, v);
                } else {
                    inProcessItemsCount++;
                }
            }

            for (Map.Entry<IndexKey, IndexValue> item : processedItems.entrySet()) {
                IndexKey k = item.getKey();
                IndexValue v = item.getValue();

                v.setProcessedMarker(currentMarker);

                levelDb.put(marshallKey(k), marshallValue(v));
            }

            log.info("list ['{}', {}] -> done ({})", now, maxEventsCount, result.size());

            if(inProcessItemsCount > 0) {
                log.warn("list ['{}', {}] -> found items in process ({})", now, maxEventsCount, inProcessItemsCount);
            }

            return result;
        } catch (Exception e) {
            log.error("list ['{}', {}] -> error", now, maxEventsCount, e);
            throw new RuntimeException("call 'list' error");
        }
    }

    private byte[] marshallKey(IndexKey key) {
        UUID id = key.getEventId();
        long expiredAt = key.getExpiredAt();

        long mostSigBits = id.getMostSignificantBits();
        long leastSigBits = id.getLeastSignificantBits();
        return ByteBuffer.allocate(8 + 8 + 8)
                .putLong(expiredAt)
                .putLong(mostSigBits)
                .putLong(leastSigBits)
                .array();
    }

    private IndexKey unmarshallKey(byte[] key) {
        ByteBuffer buffer = ByteBuffer.allocate(8 + 8 + 8).put(key);
        buffer.flip();
        long expiredAt = buffer.getLong();
        long mostSigBits = buffer.getLong();
        long leastSigBits = buffer.getLong();

        return new IndexKey(new UUID(mostSigBits, leastSigBits), expiredAt);
    }
{'fraction_non_alphanumeric': 0.12119528993556987, 'fraction_numerical': 0.002443901355254388, 'mean_word_length': 2.23499820337765, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 28, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7253204', 'n_tokens_mistral': 2618, 'n_tokens_neox': 2478, 'n_words': 530}