package com.benchmark.fixtures;

import com.benchmark.model.User;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import org.iq80.leveldb.DB;
import org.iq80.leveldb.Options;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Comparator;
import java.util.List;

import static org.iq80.leveldb.impl.Iq80DBFactory.factory;

public class LevelDbFixture implements DataFixtures {

    private DB db;
    private Kryo kryo;
    private Path dbDir;

    @Override
    public void setup(List<User> users) throws IOException {
        kryo = new Kryo();
        kryo.setRegistrationRequired(false);
        kryo.register(User.class);

        dbDir = Files.createTempDirectory("benchmark-leveldb-");
        Options options = new Options();
        options.createIfMissing(true);
        db = factory.open(dbDir.toFile(), options);

        for (User user : users) {
            db.put(toKey(user.getId()), serialize(user));
        }
    }

    @Override
    public User findById(long id) {
        byte[] bytes = db.get(toKey(id));
        return bytes == null ? null : deserialize(bytes);
    }

    @Override
    public void teardown() throws IOException {
        if (db != null) {
            db.close();
        }
        if (dbDir != null && Files.exists(dbDir)) {
            Files.walk(dbDir)
                .sorted(Comparator.reverseOrder())
                .forEach(path -> path.toFile().delete());
        }
    }

    private byte[] toKey(long id) {
        return ByteBuffer.allocate(Long.BYTES)
            .order(ByteOrder.BIG_ENDIAN)
            .putLong(id)
            .array();
    }

    private byte[] serialize(User user) {
        Output output = new Output(256, -1);
        kryo.writeObject(output, user);
        byte[] bytes = output.toBytes();
        output.close();
        return bytes;
    }

    private User deserialize(byte[] bytes) {
        try (Input input = new Input(bytes)) {
            return kryo.readObject(input, User.class);
        }
    }
}