<commit_before>package org.hive2hive.core.network.data;

import java.io.IOException;
import java.security.InvalidKeyException;
import java.security.KeyPair;
import java.security.SignatureException;

import net.tomp2p.futures.FutureGet;
import net.tomp2p.futures.FuturePut;
import net.tomp2p.futures.FutureRemove;
import net.tomp2p.p2p.Peer;
import net.tomp2p.p2p.builder.DigestBuilder;
import net.tomp2p.peers.Number160;
import net.tomp2p.peers.Number640;
import net.tomp2p.storage.Data;

import org.hive2hive.core.H2HConstants;
import org.hive2hive.core.log.H2HLogger;
import org.hive2hive.core.log.H2HLoggerFactory;
import org.hive2hive.core.network.NetworkManager;
import org.hive2hive.core.network.data.futures.FutureGetListener;
import org.hive2hive.core.network.data.futures.FuturePutListener;
import org.hive2hive.core.network.data.futures.FutureRemoveListener;

public class DataManager implements IDataManager {

	private static final H2HLogger logger = H2HLoggerFactory.getLogger(DataManager.class);

	private final NetworkManager networkManager;

	public DataManager(NetworkManager networkManager) {
		this.networkManager = networkManager;
	}

	/**
	 * Helper to get the <code>TomP2P</code> peer.
	 * 
	 * @return the current peer
	 */
	private Peer getPeer() {
		return networkManager.getConnection().getPeer();
	}

	@Override
	public boolean put(String locationKey, String contentKey, NetworkContent content, KeyPair protectionKey) {
		Number160 lKey = Number160.createHash(locationKey);
		Number160 dKey = H2HConstants.TOMP2P_DEFAULT_KEY;
		Number160 cKey = Number160.createHash(contentKey);
		FuturePut putFuture = put(lKey, dKey, cKey, content, protectionKey);
		if (putFuture == null) {
			return false;
		}

		FuturePutListener listener = new FuturePutListener(lKey, dKey, cKey, content, protectionKey, this);
		putFuture.addListener(listener);
		return listener.await();
	}

	@Override
	public boolean putUserProfileTask(String userId, Number160 contentKey, NetworkContent content,
			KeyPair protectionKey) {
		Number160 lKey = Number160.createHash(userId);
		Number160 dKey = Number160.createHash(H2HConstants.USER_PROFILE_TASK_DOMAIN);
		FuturePut putFuture = put(lKey, dKey, contentKey, content, protectionKey);
		if (putFuture == null) {
			return false;
		}

		FuturePutListener listener = new FuturePutListener(lKey, dKey, contentKey, content, protectionKey,
				this);
		putFuture.addListener(listener);
		return listener.await();
	}

	/**
	 * Putting without changing the protection key (normal case)
	 */
	public FuturePut put(Number160 locationKey, Number160 domainKey, Number160 contentKey,
			NetworkContent content, KeyPair protectionKey) {
		// hand over the same protection key twice
		return put(locationKey, domainKey, contentKey, content, protectionKey, protectionKey);
	}

	/**
	 * Putting and changing the protection key (only used for sharing, ...). The old and the new protection
	 * keys can be the same, meaning that the key is not changed
	 */
	public FuturePut put(Number160 locationKey, Number160 domainKey, Number160 contentKey,
			NetworkContent content, KeyPair oldProtectionKey, KeyPair newProtectionKey) {
		logger.debug(String
				.format("put content = '%s' location key = '%s' domain key = '%s' content key = '%s' version key = '%s' protected = '%b'",
						content.getClass().getSimpleName(), locationKey, domainKey, contentKey,
						content.getVersionKey(), newProtectionKey != null));
		try {
			Data data = new Data(content);
			data.ttlSeconds(content.getTimeToLive()).basedOn(content.getBasedOnKey());
			if (newProtectionKey == null) {
				// the content won't be protected after this put
				if (oldProtectionKey == null) {
					// previous content was not protected either
					return getPeer().put(locationKey).setData(contentKey, data).setDomainKey(domainKey)
							.setVersionKey(content.getVersionKey()).start();
				} else {
					// previous content was protected, now it should be unprotected.
					// TODO TomP2P does not support this yet
					return null;
				}
			} else {
				// the content will be protected after this put
				if (oldProtectionKey == null) {
					// the content is protected for the first time
					data.setProtectedEntry().sign(newProtectionKey);
					return getPeer().put(locationKey).setData(contentKey, data).setDomainKey(domainKey)
							.setVersionKey(content.getVersionKey()).keyPair(newProtectionKey).start();
				} else {
					// change the protection keys
					data.setProtectedEntry().sign(newProtectionKey);
					return getPeer().put(locationKey).setData(contentKey, data).setDomainKey(domainKey)
							.setVersionKey(content.getVersionKey()).keyPair(oldProtectionKey).start();
				}
			}
		} catch (IOException | InvalidKeyException | SignatureException e) {
			logger.error(String
					.format("Put failed. location key = '%s' domain key = '%s' content key = '%s' version key = '%s' exception = '%s'",
							locationKey, domainKey, contentKey, content.getVersionKey(), e.getMessage()));
			return null;
		}
	}

	// public boolean changeProtectionKey(String locationKey, String contentKey, NetworkContent content,
	// KeyPair oldProtectionKey, KeyPair newProtectionKey) {
	// Number160 lKey = Number160.createHash(locationKey);
	// Number160 cKey = Number160.createHash(contentKey);
	// Number160 dKey = H2HConstants.TOMP2P_DEFAULT_KEY;
	// FuturePut changeFuture = changeProtectionKey(lKey, cKey, dKey, content, oldProtectionKey,
	// newProtectionKey);
	// if (changeFuture == null) {
	// return false;
	// }
	//
	// FuturePutListener listener = new FuturePutListener(lKey, dKey, cKey, content, newProtectionKey, this);
	// changeFuture.addListener(listener);
	// return listener.await();
	// }

	// public FuturePut changeProtectionKey(Number160 locationKey, Number160 contentKey, Number160 domainKey,
	// NetworkContent content, KeyPair oldProtectionKey, KeyPair newProtectionKey) {
	// if (oldProtectionKey == null) {
	// logger.error("Cannot change the protection key when the original key is not given");
	// return null;
	// }
	//
	// logger.debug(String
	// .format("change protection key content = '%s' location key = '%s' domain key = '%s' content key = '%s' version key = '%s'",
	// content.getClass().getSimpleName(), locationKey, domainKey, contentKey,
	// content.getVersionKey()));
	//
	// try {
	// Data data = new Data(content);
	// data.ttlSeconds(content.getTimeToLive()).basedOn(content.getBasedOnKey());
	// if (newProtectionKey == null) {
	// return getPeer().put(locationKey).setData(contentKey, data).setDomainKey(domainKey)
	// .setVersionKey(content.getVersionKey()).start();
	// } else {
	// data.setProtectedEntry().sign(newProtectionKey);
	// return getPeer().put(locationKey).setData(contentKey, data).setDomainKey(domainKey)
	// .setVersionKey(content.getVersionKey()).keyPair(oldProtectionKey).start();
	// }
	// } catch (IOException | InvalidKeyException | SignatureException e) {
	// logger.error(String
	// .format("Chanign protection key failed. location key = '%s' domain key = '%s' content key = '%s' version key = '%s' exception = '%s'",
	// locationKey, domainKey, contentKey, content.getVersionKey(), e.getMessage()));
	// return null;
	// }
	//
	// }

	@Override
	public NetworkContent get(String locationKey, String contentKey) {
		Number160 lKey = Number160.createHash(locationKey);
		Number160 dKey = H2HConstants.TOMP2P_DEFAULT_KEY;
		Number160 cKey = Number160.createHash(contentKey);

		FutureGet futureGet = get(lKey, dKey, cKey);
		FutureGetListener listener = new FutureGetListener(lKey, dKey, cKey, this);
		futureGet.addListener(listener);
		return listener.awaitAndGet();
	}

	public NetworkContent get(String locationKey, String contentKey, Number160 versionKey) {
		Number160 lKey = Number160.createHash(locationKey);
		Number160 dKey = H2HConstants.TOMP2P_DEFAULT_KEY;
		Number160 cKey = Number160.createHash(contentKey);

		FutureGet futureGet = get(lKey, dKey, cKey, versionKey);
		FutureGetListener listener = new FutureGetListener(lKey, dKey, cKey, versionKey, this);
		futureGet.addListener(listener);
		return listener.awaitAndGet();
	}

	@Override
	public NetworkContent getUserProfileTask(String userId) {
		Number160 lKey = Number160.createHash(userId);
		Number160 dKey = Number160.createHash(H2HConstants.USER_PROFILE_TASK_DOMAIN);
<commit_msg>Two steps for ttl and basedOn (easier to uncomment) + old code deletion<commit_after>package org.hive2hive.core.network.data;

import java.io.IOException;
import java.security.InvalidKeyException;
import java.security.KeyPair;
import java.security.SignatureException;

import net.tomp2p.futures.FutureGet;
import net.tomp2p.futures.FuturePut;
import net.tomp2p.futures.FutureRemove;
import net.tomp2p.p2p.Peer;
import net.tomp2p.p2p.builder.DigestBuilder;
import net.tomp2p.peers.Number160;
import net.tomp2p.peers.Number640;
import net.tomp2p.storage.Data;

import org.hive2hive.core.H2HConstants;
import org.hive2hive.core.log.H2HLogger;
import org.hive2hive.core.log.H2HLoggerFactory;
import org.hive2hive.core.network.NetworkManager;
import org.hive2hive.core.network.data.futures.FutureGetListener;
import org.hive2hive.core.network.data.futures.FuturePutListener;
import org.hive2hive.core.network.data.futures.FutureRemoveListener;

public class DataManager implements IDataManager {

	private static final H2HLogger logger = H2HLoggerFactory.getLogger(DataManager.class);

	private final NetworkManager networkManager;

	public DataManager(NetworkManager networkManager) {
		this.networkManager = networkManager;
	}

	/**
	 * Helper to get the <code>TomP2P</code> peer.
	 * 
	 * @return the current peer
	 */
	private Peer getPeer() {
		return networkManager.getConnection().getPeer();
	}

	@Override
	public boolean put(String locationKey, String contentKey, NetworkContent content, KeyPair protectionKey) {
		Number160 lKey = Number160.createHash(locationKey);
		Number160 dKey = H2HConstants.TOMP2P_DEFAULT_KEY;
		Number160 cKey = Number160.createHash(contentKey);
		FuturePut putFuture = put(lKey, dKey, cKey, content, protectionKey);
		if (putFuture == null) {
			return false;
		}

		FuturePutListener listener = new FuturePutListener(lKey, dKey, cKey, content, protectionKey, this);
		putFuture.addListener(listener);
		return listener.await();
	}

	@Override
	public boolean putUserProfileTask(String userId, Number160 contentKey, NetworkContent content,
			KeyPair protectionKey) {
		Number160 lKey = Number160.createHash(userId);
		Number160 dKey = Number160.createHash(H2HConstants.USER_PROFILE_TASK_DOMAIN);
		FuturePut putFuture = put(lKey, dKey, contentKey, content, protectionKey);
		if (putFuture == null) {
			return false;
		}

		FuturePutListener listener = new FuturePutListener(lKey, dKey, contentKey, content, protectionKey,
				this);
		putFuture.addListener(listener);
		return listener.await();
	}

	/**
	 * Putting without changing the protection key (normal case)
	 */
	public FuturePut put(Number160 locationKey, Number160 domainKey, Number160 contentKey,
			NetworkContent content, KeyPair protectionKey) {
		// hand over the same protection key twice
		return put(locationKey, domainKey, contentKey, content, protectionKey, protectionKey);
	}

	/**
	 * Putting and changing the protection key (only used for sharing, ...). The old and the new protection
	 * keys can be the same, meaning that the key is not changed
	 */
	public FuturePut put(Number160 locationKey, Number160 domainKey, Number160 contentKey,
			NetworkContent content, KeyPair oldProtectionKey, KeyPair newProtectionKey) {
		logger.debug(String
				.format("put content = '%s' location key = '%s' domain key = '%s' content key = '%s' version key = '%s' protected = '%b'",
						content.getClass().getSimpleName(), locationKey, domainKey, contentKey,
						content.getVersionKey(), newProtectionKey != null));
		try {
			Data data = new Data(content);
			data.ttlSeconds(content.getTimeToLive());
			data.basedOn(content.getBasedOnKey());
			if (newProtectionKey == null) {
				// the content won't be protected after this put
				if (oldProtectionKey == null) {
					// previous content was not protected either
					return getPeer().put(locationKey).setData(contentKey, data).setDomainKey(domainKey)
							.setVersionKey(content.getVersionKey()).start();
				} else {
					// previous content was protected, now it should be unprotected.
					// TODO TomP2P does not support this yet
					return null;
				}
			} else {
				// the content will be protected after this put
				if (oldProtectionKey == null) {
					// the content is protected for the first time
					data.setProtectedEntry().sign(newProtectionKey);
					return getPeer().put(locationKey).setData(contentKey, data).setDomainKey(domainKey)
							.setVersionKey(content.getVersionKey()).keyPair(newProtectionKey).start();
				} else {
					// change the protection keys
					data.setProtectedEntry().sign(newProtectionKey);
					return getPeer().put(locationKey).setData(contentKey, data).setDomainKey(domainKey)
							.setVersionKey(content.getVersionKey()).keyPair(oldProtectionKey).start();
				}
			}
		} catch (IOException | InvalidKeyException | SignatureException e) {
			logger.error(String
					.format("Put failed. location key = '%s' domain key = '%s' content key = '%s' version key = '%s' exception = '%s'",
							locationKey, domainKey, contentKey, content.getVersionKey(), e.getMessage()));
			return null;
		}
	}

	@Override
	public NetworkContent get(String locationKey, String contentKey) {
		Number160 lKey = Number160.createHash(locationKey);
		Number160 dKey = H2HConstants.TOMP2P_DEFAULT_KEY;
		Number160 cKey = Number160.createHash(contentKey);

		FutureGet futureGet = get(lKey, dKey, cKey);
		FutureGetListener listener = new FutureGetListener(lKey, dKey, cKey, this);
		futureGet.addListener(listener);
		return listener.awaitAndGet();
	}

	public NetworkContent get(String locationKey, String contentKey, Number160 versionKey) {
		Number160 lKey = Number160.createHash(locationKey);
		Number160 dKey = H2HConstants.TOMP2P_DEFAULT_KEY;
		Number160 cKey = Number160.createHash(contentKey);

		FutureGet futureGet = get(lKey, dKey, cKey, versionKey);
		FutureGetListener listener = new FutureGetListener(lKey, dKey, cKey, versionKey, this);
		futureGet.addListener(listener);
		return listener.awaitAndGet();
	}

	@Override
	public NetworkContent getUserProfileTask(String userId) {
		Number160 lKey = Number160.createHash(userId);
		Number160 dKey = Number160.createHash(H2HConstants.USER_PROFILE_TASK_DOMAIN);

{'fraction_non_alphanumeric': 0.11039988987542157, 'fraction_numerical': 0.01996007984031936, 'mean_word_length': 4.8050339592489015, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, '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': '6988937', 'n_tokens_mistral': 4927, 'n_tokens_neox': 4396, 'n_words': 1062}