Storage: typed heads

3 files changed, 352 insertions, 14 deletions

diff --git a/include/erebos/storage.h b/include/erebos/storage.h
index 09dc481..57304e1 100644
--- a/include/erebos/storage.h
+++ b/include/erebos/storage.h
@@ -7,6 +7,7 @@
 #include <array>
 #include <cstring>
 #include <filesystem>
+#include <functional>
 #include <future>
 #include <memory>
 #include <optional>
@@ -31,6 +32,7 @@ typedef ObjectT<PartialStorage> PartialObject;
 class Blob;
 
 template<typename T> class Stored;
+template<typename T> class Head;
 
 class PartialStorage
 {
@@ -87,11 +89,24 @@ public:
 
 	template<typename T> Stored<T> store(const T &) const;
 
+	template<typename T> std::optional<Head<T>> head(UUID id) const;
+	template<typename T> std::vector<Head<T>> heads() const;
+	template<typename T> Head<T> storeHead(const T &) const;
+	template<typename T> Head<T> storeHead(const Stored<T> &) const;
+
 	void storeKey(Ref pubref, const std::vector<uint8_t> &) const;
 	std::optional<std::vector<uint8_t>> loadKey(Ref pubref) const;
 
 protected:
+	template<typename T> friend class Head;
+
 	Storage(const std::shared_ptr<const Priv> p): PartialStorage(p) {}
+
+	std::optional<Ref> headRef(UUID type, UUID id) const;
+	std::vector<std::tuple<UUID, Ref>> headRefs(UUID type) const;
+	static UUID storeHead(UUID type, const Ref & ref);
+	static bool replaceHead(UUID type, UUID id, const Ref & old, const Ref & ref);
+	static std::optional<Ref> updateHead(UUID type, UUID id, const Ref & old, const std::function<Ref(const Ref &)> &);
 };
 
 class Digest
@@ -316,12 +331,14 @@ class Stored
 {
 	Stored(Ref ref, std::future<T> && val): mref(ref), mval(std::move(val)) {}
 	friend class Storage;
+	friend class Head<T>;
 public:
 	Stored(const Stored &) = default;
 	Stored(Stored &&) = default;
 	Stored & operator=(const Stored &) = default;
 	Stored & operator=(Stored &&) = default;
 
+	Stored(const Ref &);
 	static Stored<T> load(const Ref &);
 	Ref store(const Storage &) const;
 
@@ -360,11 +377,17 @@ Stored<T> Storage::store(const T & val) const
 }
 
 template<typename T>
+Stored<T>::Stored(const Ref & ref):
+	mref(ref),
+	mval(std::async(std::launch::deferred, [ref] {
+		return T::load(ref);
+	}))
+{}
+
+template<typename T>
 Stored<T> Stored<T>::load(const Ref & ref)
 {
-	return Stored(ref, std::async(std::launch::deferred, [ref] {
-		return T::load(ref);
-	}));
+	return Stored(ref);
 }
 
 template<typename T>
@@ -436,6 +459,80 @@ void filterAncestors(std::vector<Stored<T>> & xs)
 	}
 }
 
+template<class T>
+class Head
+{
+	Head(UUID id, Ref ref, std::future<T> && val):
+		mid(id), mstored(ref, std::move(val)) {}
+	friend class Storage;
+public:
+	Head(UUID id, Ref ref): mid(id), mstored(ref) {}
+
+	const T & operator*() const { return *mstored; }
+	const T * operator->() const { return &(*mstored); }
+
+	std::vector<Stored<T>> previous() const;
+	bool precedes(const Stored<T> &) const;
+
+	UUID id() const { return mid; }
+	const Stored<T> & stored() const { return mstored; }
+	const Ref & ref() const { return mstored.ref(); }
+
+	std::optional<Head<T>> update(const std::function<Stored<T>(const Stored<T> &)> &) const;
+
+private:
+	UUID mid;
+	Stored<T> mstored;
+};
+
+template<typename T>
+std::optional<Head<T>> Storage::head(UUID id) const
+{
+	if (auto ref = headRef(T::headTypeId, id))
+		return Head<T>(id, *ref);
+	return std::nullopt;
+}
+
+template<typename T>
+std::vector<Head<T>> Storage::heads() const
+{
+	std::vector<Head<T>> res;
+	for (const auto & x : headRefs(T::headTypeId))
+		res.emplace_back(std::get<UUID>(x), std::get<Ref>(x));
+	return res;
+}
+
+template<typename T>
+Head<T> Storage::storeHead(const T & val) const
+{
+	auto ref = val.store(*this);
+	auto id = storeHead(T::headTypeId, ref);
+	return Head(id, ref, std::async(std::launch::deferred, [val] {
+		return val;
+	}));
+}
+
+template<typename T>
+Head<T> Storage::storeHead(const Stored<T> & val) const
+{
+	auto id = storeHead(T::headTypeId, val.ref());
+	return Head(id, val.ref(), val.mval);
+}
+
+template<typename T>
+std::optional<Head<T>> Head<T>::update(const std::function<Stored<T>(const Stored<T> &)> & f) const
+{
+	auto res = Storage::updateHead(T::headTypeId, mid, ref(), [&f, this](const Ref & r) {
+		return f(r == ref() ? stored() : Stored<T>::load(r)).ref();
+	});
+
+	if (!res)
+		return std::nullopt;
+	if (*res == ref())
+		return *this;
+	return Head<T>(mid, *res);
+}
+
 }
 
 namespace std
diff --git a/src/storage.cpp b/src/storage.cpp
index 16662df..b5bff8c 100644
--- a/src/storage.cpp
+++ b/src/storage.cpp
@@ -18,6 +18,7 @@ using namespace erebos;
 
 using std::array;
 using std::copy;
+using std::get;
 using std::holds_alternative;
 using std::ifstream;
 using std::is_same_v;
@@ -30,7 +31,6 @@ using std::runtime_error;
 using std::shared_ptr;
 using std::string;
 using std::to_string;
-using std::tuple;
 
 FilesystemStorage::FilesystemStorage(const fs::path & path):
 	root(path)
@@ -125,16 +125,7 @@ void FilesystemStorage::storeBytes(const Digest & digest, const vector<uint8_t>
 	auto lock = path;
 	lock += ".lock";
 
-	fs::create_directories(path.parent_path());
-
-	// No way to use open exclusively in c++ stdlib
-	FILE *f = nullptr;
-	for (int i = 0; i < 10; i++) {
-		f = fopen(lock.c_str(), "wbxe");
-		if (f || errno != EEXIST)
-			break;
-		std::this_thread::sleep_for(std::chrono::milliseconds(100));
-	}
+	FILE * f = openLockFile(lock);
 	if (fs::exists(path)) {
 		if (f) {
 			fclose(f);
@@ -171,6 +162,81 @@ void FilesystemStorage::storeBytes(const Digest & digest, const vector<uint8_t>
 	fs::rename(lock, path);
 }
 
+optional<Digest> FilesystemStorage::headRef(UUID type, UUID id) const
+{
+	ifstream fin(headPath(type, id));
+	if (!fin)
+		return nullopt;
+
+	string sdgst;
+	fin >> sdgst;
+	return Digest(sdgst);
+}
+
+vector<tuple<UUID, Digest>> FilesystemStorage::headRefs(UUID type) const
+{
+	vector<tuple<UUID, Digest>> res;
+	string stype(type);
+	fs::path ptype(stype.begin(), stype.end());
+	try {
+		for (const auto & p : fs::directory_iterator(root/"heads"/ptype))
+			if (auto u = UUID::fromString(p.path().filename())) {
+				ifstream fin(p.path());
+				if (fin) {
+					string sdgst;
+					fin >> sdgst;
+					res.emplace_back(*u, Digest(sdgst));
+				}
+			}
+	} catch (const fs::filesystem_error & e) {
+		if (e.code() == std::errc::no_such_file_or_directory)
+			return {};
+		throw e;
+	}
+	return res;
+}
+
+UUID FilesystemStorage::storeHead(UUID type, const Digest & dgst)
+{
+	auto id = UUID::generate();
+	auto path = headPath(type, id);
+	fs::create_directories(path.parent_path());
+	ofstream fout(path);
+	if (!fout)
+		throw runtime_error("failed to open head file");
+
+	fout << string(dgst) << '\n';
+	return id;
+}
+
+bool FilesystemStorage::replaceHead(UUID type, UUID id, const Digest & old, const Digest & dgst)
+{
+	auto path = headPath(type, id);
+	auto lock = path;
+	lock += ".lock";
+	FILE * f = openLockFile(lock);
+	if (!f)
+		throw runtime_error(("failed to lock head file " + string(path) +
+					": " + string(strerror(errno))).c_str());
+
+	string scur;
+	ifstream fin(path);
+	fin >> scur;
+	fin.close();
+	Digest cur(scur);
+
+	if (cur != old) {
+		fclose(f);
+		unlink(lock.c_str());
+		return false;
+	}
+
+	fprintf(f, "%s\n", string(dgst).c_str());
+	fclose(f);
+	fs::rename(lock, path);
+	return true;
+}
+
 optional<vector<uint8_t>> FilesystemStorage::loadKey(const Digest & pubref) const
 {
 	fs::path path = keyPath(pubref);
@@ -201,12 +267,37 @@ fs::path FilesystemStorage::objectPath(const Digest & digest) const
 		fs::path(name.begin() + 2, name.end());
 }
 
+fs::path FilesystemStorage::headPath(UUID type, UUID id) const
+{
+	string stype(type), sid(id);
+	return root/"heads"/
+		fs::path(stype.begin(), stype.end())/
+		fs::path(sid.begin(), sid.end());
+}
+
 fs::path FilesystemStorage::keyPath(const Digest & digest) const
 {
 	string name(digest);
 	return root/"keys"/fs::path(name.begin(), name.end());
 }
 
+FILE * FilesystemStorage::openLockFile(const fs::path & path) const
+{
+	fs::create_directories(path.parent_path());
+
+	// No way to use open exclusively in c++ stdlib
+	FILE *f = nullptr;
+	for (int i = 0; i < 10; i++) {
+		f = fopen(path.c_str(), "wbxe");
+		if (f || errno != EEXIST)
+			break;
+		std::this_thread::sleep_for(std::chrono::milliseconds(100));
+	}
+
+	return f;
+}
+
+
 bool MemoryStorage::contains(const Digest & digest) const
 {
 	return storage.find(digest) != storage.end();
@@ -225,6 +316,57 @@ void MemoryStorage::storeBytes(const Digest & digest, const vector<uint8_t> & co
 	storage.emplace(digest, content);
 }
 
+optional<Digest> MemoryStorage::headRef(UUID type, UUID id) const
+{
+	auto it = heads.find(type);
+	if (it == heads.end())
+		return nullopt;
+
+	for (const auto & x : it->second)
+		if (get<UUID>(x) == id)
+			return get<Digest>(x);
+
+	return nullopt;
+}
+
+vector<tuple<UUID, Digest>> MemoryStorage::headRefs(UUID type) const
+{
+	auto it = heads.find(type);
+	if (it != heads.end())
+		return it->second;
+	return {};
+}
+
+UUID MemoryStorage::storeHead(UUID type, const Digest & dgst)
+{
+	auto id = UUID::generate();
+	auto it = heads.find(type);
+	if (it == heads.end())
+		heads[type] = { { id, dgst } };
+	else
+		it->second.emplace_back(id, dgst);
+	return id;
+}
+
+bool MemoryStorage::replaceHead(UUID type, UUID id, const Digest & old, const Digest & dgst)
+{
+	auto it = heads.find(type);
+	if (it == heads.end())
+		return false;
+
+	for (auto & x : it->second)
+		if (get<UUID>(x) == id) {
+			if (get<Digest>(x) == old) {
+				get<Digest>(x) = dgst;
+				return true;
+			} else {
+				return false;
+			}
+		}
+
+	return false;
+}
+
 optional<vector<uint8_t>> MemoryStorage::loadKey(const Digest & digest) const
 {
 	auto it = keys.find(digest);
@@ -258,6 +400,42 @@ void ChainStorage::storeBytes(const Digest & digest, const vector<uint8_t> & con
 	storage->storeBytes(digest, content);
 }
 
+optional<Digest> ChainStorage::headRef(UUID type, UUID id) const
+{
+	if (auto res = storage->headRef(type, id))
+		return res;
+	if (parent)
+		return parent->headRef(type, id);
+	return nullopt;
+}
+
+vector<tuple<UUID, Digest>> ChainStorage::headRefs(UUID type) const
+{
+	auto res = storage->headRefs(type);
+	if (parent)
+		for (auto x : parent->headRefs(type)) {
+			bool add = true;
+			for (const auto & y : res)
+				if (get<UUID>(y) == get<UUID>(x)) {
+					add = false;
+					break;
+				}
+			if (add)
+				res.push_back(x);
+		}
+	return res;
+}
+
+UUID ChainStorage::storeHead(UUID type, const Digest & dgst)
+{
+	return storage->storeHead(type, dgst);
+}
+
+bool ChainStorage::replaceHead(UUID type, UUID id, const Digest & old, const Digest & dgst)
+{
+	return storage->replaceHead(type, id, old, dgst);
+}
+
 optional<vector<uint8_t>> ChainStorage::loadKey(const Digest & digest) const
 {
 	if (auto res = storage->loadKey(digest))
@@ -453,6 +631,44 @@ optional<vector<uint8_t>> Storage::loadKey(Ref pubref) const
 	return p->backend->loadKey(pubref.digest());
 }
 
+optional<Ref> Storage::headRef(UUID type, UUID id) const
+{
+	if (auto dgst = p->backend->headRef(type, id))
+		return ref(*dgst);
+	return nullopt;
+}
+
+vector<tuple<UUID, Ref>> Storage::headRefs(UUID type) const
+{
+	vector<tuple<UUID, Ref>> res;
+	for (auto x : p->backend->headRefs(type))
+		if (auto r = ref(get<Digest>(x)))
+			res.emplace_back(get<UUID>(x), *r);
+	return res;
+}
+
+UUID Storage::storeHead(UUID type, const Ref & ref)
+{
+	return ref.storage().p->backend->storeHead(type, ref.digest());
+}
+
+bool Storage::replaceHead(UUID type, UUID id, const Ref & old, const Ref & ref)
+{
+	return ref.storage().p->backend->replaceHead(type, id, old.digest(), ref.digest());
+}
+
+optional<Ref> Storage::updateHead(UUID type, UUID id, const Ref & old, const std::function<Ref(const Ref &)> & f)
+{
+	Ref r = f(old);
+	if (replaceHead(type, id, old, r))
+		return r;
+
+	if (auto cur = old.storage().headRef(type, id))
+		return updateHead(type, id, *cur, f);
+	else
+		return nullopt;
+}
+
 
 Digest::Digest(const string & str)
 {
diff --git a/src/storage.h b/src/storage.h
index 1635d40..18ac1ad 100644
--- a/src/storage.h
+++ b/src/storage.h
@@ -13,6 +13,7 @@ using std::shared_ptr;
 using std::unique_ptr;
 using std::unordered_map;
 using std::unordered_set;
+using std::tuple;
 using std::variant;
 using std::vector;
 
@@ -29,6 +30,11 @@ public:
 	virtual optional<vector<uint8_t>> loadBytes(const Digest &) const = 0;
 	virtual void storeBytes(const Digest &, const vector<uint8_t> &) = 0;
 
+	virtual optional<Digest> headRef(UUID type, UUID id) const = 0;
+	virtual vector<tuple<UUID, Digest>> headRefs(UUID type) const = 0;
+	virtual UUID storeHead(UUID type, const Digest & dgst) = 0;
+	virtual bool replaceHead(UUID type, UUID id, const Digest & old, const Digest & dgst) = 0;
+
 	virtual optional<vector<uint8_t>> loadKey(const Digest &) const = 0;
 	virtual void storeKey(const Digest &, const vector<uint8_t> &) = 0;
 };
@@ -44,6 +50,11 @@ public:
 	virtual optional<vector<uint8_t>> loadBytes(const Digest &) const override;
 	virtual void storeBytes(const Digest &, const vector<uint8_t> &) override;
 
+	virtual optional<Digest> headRef(UUID type, UUID id) const override;
+	virtual vector<tuple<UUID, Digest>> headRefs(UUID type) const override;
+	virtual UUID storeHead(UUID type, const Digest & dgst) override;
+	virtual bool replaceHead(UUID type, UUID id, const Digest & old, const Digest & dgst) override;
+
 	virtual optional<vector<uint8_t>> loadKey(const Digest &) const override;
 	virtual void storeKey(const Digest &, const vector<uint8_t> &) override;
 
@@ -51,8 +62,11 @@ private:
 	static constexpr size_t CHUNK = 16384;
 
 	fs::path objectPath(const Digest &) const;
+	fs::path headPath(UUID id, UUID type) const;
 	fs::path keyPath(const Digest &) const;
 
+	FILE * openLockFile(const fs::path & path) const;
+
 	fs::path root;
 };
 
@@ -67,11 +81,17 @@ public:
 	virtual optional<vector<uint8_t>> loadBytes(const Digest &) const override;
 	virtual void storeBytes(const Digest &, const vector<uint8_t> &) override;
 
+	virtual optional<Digest> headRef(UUID type, UUID id) const override;
+	virtual vector<tuple<UUID, Digest>> headRefs(UUID type) const override;
+	virtual UUID storeHead(UUID type, const Digest & dgst) override;
+	virtual bool replaceHead(UUID type, UUID id, const Digest & old, const Digest & dgst) override;
+
 	virtual optional<vector<uint8_t>> loadKey(const Digest &) const override;
 	virtual void storeKey(const Digest &, const vector<uint8_t> &) override;
 
 private:
 	unordered_map<Digest, vector<uint8_t>> storage;
+	unordered_map<UUID, vector<tuple<UUID, Digest>>> heads;
 	unordered_map<Digest, vector<uint8_t>> keys;
 };
 
@@ -89,6 +109,11 @@ public:
 	virtual optional<vector<uint8_t>> loadBytes(const Digest &) const override;
 	virtual void storeBytes(const Digest &, const vector<uint8_t> &) override;
 
+	virtual optional<Digest> headRef(UUID type, UUID id) const override;
+	virtual vector<tuple<UUID, Digest>> headRefs(UUID type) const override;
+	virtual UUID storeHead(UUID type, const Digest & dgst) override;
+	virtual bool replaceHead(UUID type, UUID id, const Digest & old, const Digest & dgst) override;
+
 	virtual optional<vector<uint8_t>> loadKey(const Digest &) const override;
 	virtual void storeKey(const Digest &, const vector<uint8_t> &) override;