CoreStorage saves Topic, Repo, Overlay and Commit

pull/19/head
Niko PLP 7 months ago
parent 2aa996f01e
commit 508c7bf177
  1. 7
      ng-broker/src/rocksdb_server_storage.rs
  2. 32
      ng-broker/src/server_broker.rs
  3. 2
      ng-broker/src/server_storage/admin/account.rs
  4. 2
      ng-broker/src/server_storage/admin/invitation.rs
  5. 2
      ng-broker/src/server_storage/admin/wallet.rs
  6. 133
      ng-broker/src/server_storage/core/commit.rs
  7. 3
      ng-broker/src/server_storage/core/mod.rs
  8. 277
      ng-broker/src/server_storage/core/overlay.rs
  9. 12
      ng-broker/src/server_storage/core/repo.rs
  10. 41
      ng-broker/src/server_storage/core/topic.rs
  11. 98
      ng-repo/src/kcv_storage.rs
  12. 2
      ng-verifier/src/user_storage/branch.rs
  13. 2
      ng-verifier/src/user_storage/repo.rs

@ -126,14 +126,17 @@ impl RocksDbServerStorage {
// check unicity of class prefixes, by storage // check unicity of class prefixes, by storage
#[cfg(debug_assertions)] #[cfg(debug_assertions)]
{ {
// TODO: refactor the wallet and accounts with Class and the new OKM mechanism, then include them uncomment the following lines
//log_debug!("CHECKING..."); //log_debug!("CHECKING...");
// wallet_storage.add_class(&Wallet::CLASS); // wallet_storage.add_class(&Wallet::CLASS);
// wallet_storage.check_prefixes(); // wallet_storage.check_prefixes();
// accounts_storage.add_class(&Account::CLASS); // accounts_storage.add_class(&Account::CLASS);
// accounts_storage.add_class(&Invitation::CLASS); // accounts_storage.add_class(&Invitation::CLASS);
// accounts_storage.check_prefixes(); // accounts_storage.check_prefixes();
core_storage.add_class(&Topic::CLASS); core_storage.add_class(&TopicStorage::CLASS);
core_storage.add_class(&RepoOKM::CLASS); core_storage.add_class(&RepoHashStorage::CLASS);
core_storage.add_class(&OverlayStorage::CLASS);
core_storage.add_class(&CommitStorage::CLASS);
core_storage.check_prefixes(); core_storage.check_prefixes();
} }

@ -18,6 +18,7 @@ use ng_repo::{
errors::{NgError, ProtocolError, ServerError}, errors::{NgError, ProtocolError, ServerError},
types::*, types::*,
}; };
use serde::{Deserialize, Serialize};
use crate::rocksdb_server_storage::RocksDbServerStorage; use crate::rocksdb_server_storage::RocksDbServerStorage;
@ -43,21 +44,38 @@ pub struct RepoInfo {
pub topics: HashSet<TopicId>, pub topics: HashSet<TopicId>,
} }
pub struct OverlayInfo { #[derive(Clone, Debug, Serialize, Deserialize)]
inner: Option<OverlayId>, pub struct EventInfo {
pub event: Event,
pub blocks: Vec<BlockId>,
}
overlay_topic: Option<TopicId>, pub struct CommitInfo {
pub event: Option<EventInfo>,
pub home_pinned: bool,
pub acks: HashSet<ObjectId>,
pub deps: HashSet<ObjectId>,
pub futures: HashSet<ObjectId>,
pub files: HashSet<ObjectId>,
}
topics: HashMap<TopicId, TopicInfo>, #[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)]
pub enum OverlayType {
Outer(OverlayId), // the ID of the inner overlay corresponding to this outer.
Inner(OverlayId), // the ID of the outer overlay corresponding to the inner
InnerOnly,
}
repos: HashMap<RepoHash, RepoInfo>, pub struct OverlayInfo {
pub overlay_type: OverlayType,
pub overlay_topic: Option<TopicId>,
pub topics: HashMap<TopicId, TopicInfo>,
pub repos: HashMap<RepoHash, RepoInfo>,
} }
pub struct ServerBroker { pub struct ServerBroker {
storage: RocksDbServerStorage, storage: RocksDbServerStorage,
overlays: HashMap<OverlayId, OverlayInfo>, overlays: HashMap<OverlayId, OverlayInfo>,
inner_overlays: HashMap<OverlayId, Option<OverlayId>>, inner_overlays: HashMap<OverlayId, Option<OverlayId>>,
} }

@ -7,7 +7,7 @@
// notice may not be copied, modified, or distributed except // notice may not be copied, modified, or distributed except
// according to those terms. // according to those terms.
//! User account OKM (Object Key/Col/Value Mapping) //! User account Storage (Object Key/Col/Value Mapping)
use std::collections::hash_map::DefaultHasher; use std::collections::hash_map::DefaultHasher;
use std::hash::Hash; use std::hash::Hash;

@ -7,7 +7,7 @@
// notice may not be copied, modified, or distributed except // notice may not be copied, modified, or distributed except
// according to those terms. // according to those terms.
//! User account OKM (Object Key/Col/Value Mapping) //! User account Storage (Object Key/Col/Value Mapping)
use std::collections::hash_map::DefaultHasher; use std::collections::hash_map::DefaultHasher;
use std::hash::Hash; use std::hash::Hash;

@ -7,7 +7,7 @@
// notice may not be copied, modified, or distributed except // notice may not be copied, modified, or distributed except
// according to those terms. // according to those terms.
//! Broker Wallet OKM (Object Key/Col/Value Mapping), persists to storage all the SymKeys needed to open other storages //! Broker Wallet Storage (Object Key/Col/Value Mapping), persists to storage all the SymKeys needed to open other storages
use ng_net::types::*; use ng_net::types::*;
use ng_repo::errors::StorageError; use ng_repo::errors::StorageError;

@ -0,0 +1,133 @@
// Copyright (c) 2022-2024 Niko Bonnieure, Par le Peuple, NextGraph.org developers
// All rights reserved.
// Licensed under the Apache License, Version 2.0
// <LICENSE-APACHE2 or http://www.apache.org/licenses/LICENSE-2.0>
// or the MIT license <LICENSE-MIT or http://opensource.org/licenses/MIT>,
// at your option. All files in the project carrying such
// notice may not be copied, modified, or distributed except
// according to those terms.
//! Commit Storage (Object Key/Col/Value Mapping)
use std::collections::HashMap;
use std::collections::HashSet;
use ng_net::types::*;
use ng_repo::errors::StorageError;
use ng_repo::kcv_storage::*;
use ng_repo::types::*;
use serde_bare::to_vec;
use crate::server_broker::CommitInfo;
use crate::server_broker::EventInfo;
pub struct CommitStorage<'a> {
key: Vec<u8>,
event: ExistentialValue<Option<EventInfo>>,
storage: &'a dyn KCVStorage,
}
impl<'a> IModel for CommitStorage<'a> {
fn key(&self) -> &Vec<u8> {
&self.key
}
fn storage(&self) -> &dyn KCVStorage {
self.storage
}
fn class(&self) -> &Class {
&Self::CLASS
}
fn existential(&mut self) -> Option<&mut dyn IExistentialValue> {
Some(&mut self.event)
}
}
impl<'a> CommitStorage<'a> {
const PREFIX: u8 = b'e';
// Topic properties
pub const EVENT: ExistentialValueColumn = ExistentialValueColumn::new(b'e');
pub const HOME_PINNED: SingleValueColumn<Self, bool> = SingleValueColumn::new(b'p');
// Commit -> Acks
pub const ACKS: MultiValueColumn<Self, ObjectId> = MultiValueColumn::new(b'a');
// Commit -> Deps
pub const DEPS: MultiValueColumn<Self, ObjectId> = MultiValueColumn::new(b'd');
// Commit -> Files
pub const FILES: MultiValueColumn<Self, ObjectId> = MultiValueColumn::new(b'f');
// Commit -> Causal future commits
pub const FUTURES: MultiValueColumn<Self, ObjectId> = MultiValueColumn::new(b'c');
pub const CLASS: Class<'a> = Class::new(
"Commit",
Some(Self::PREFIX),
Some(&Self::EVENT),
&[&Self::HOME_PINNED as &dyn ISingleValueColumn],
&[
&Self::ACKS as &dyn IMultiValueColumn,
&Self::DEPS,
&Self::FILES,
&Self::FUTURES,
],
);
pub fn new(id: &ObjectId, overlay: &OverlayId, storage: &'a dyn KCVStorage) -> Self {
let mut key: Vec<u8> = Vec::with_capacity(33 + 33);
key.append(&mut to_vec(overlay).unwrap());
key.append(&mut to_vec(id).unwrap());
CommitStorage {
key,
event: ExistentialValue::<Option<EventInfo>>::new(),
storage,
}
}
pub fn load(
id: &ObjectId,
overlay: &OverlayId,
storage: &'a dyn KCVStorage,
) -> Result<CommitInfo, StorageError> {
let mut opening = CommitStorage::new(id, overlay, storage);
let props = opening.load_props()?;
let existential = col(&Self::EVENT, &props)?;
opening.event.set(&existential)?;
Ok(CommitInfo {
event: existential,
home_pinned: col(&Self::HOME_PINNED, &props).unwrap_or(false),
acks: Self::ACKS.get_all(&mut opening)?,
deps: Self::DEPS.get_all(&mut opening)?,
files: Self::FILES.get_all(&mut opening)?,
futures: Self::FUTURES.get_all(&mut opening)?,
})
}
pub fn open(
id: &ObjectId,
overlay: &OverlayId,
storage: &'a dyn KCVStorage,
) -> Result<CommitStorage<'a>, StorageError> {
let mut opening = CommitStorage::new(id, overlay, storage);
opening.check_exists()?;
Ok(opening)
}
pub fn create(
id: &ObjectId,
overlay: &OverlayId,
event: &Option<EventInfo>,
storage: &'a mut dyn KCVStorage,
) -> Result<CommitStorage<'a>, StorageError> {
let mut creating = CommitStorage::new(id, overlay, storage);
if creating.exists() {
return Err(StorageError::AlreadyExists);
}
creating.event.set(event)?;
ExistentialValue::save(&creating, event)?;
Ok(creating)
}
pub fn event(&mut self) -> &Option<EventInfo> {
self.event.get().unwrap()
}
}

@ -9,3 +9,6 @@ pub use topic::*;
pub mod repo; pub mod repo;
pub use repo::*; pub use repo::*;
pub mod commit;
pub use commit::*;

@ -7,229 +7,112 @@
// notice may not be copied, modified, or distributed except // notice may not be copied, modified, or distributed except
// according to those terms. // according to those terms.
//! Overlay OKM (Object Key/Col/Value Mapping) //! Overlay Storage (Object Key/Col/Value Mapping)
use std::collections::HashMap;
use std::collections::HashSet;
use ng_net::types::*; use ng_net::types::*;
use ng_repo::errors::StorageError; use ng_repo::errors::StorageError;
use ng_repo::kcv_storage::KCVStorage; use ng_repo::kcv_storage::*;
use ng_repo::types::*; use ng_repo::types::*;
use ng_repo::utils::now_timestamp;
use serde::{Deserialize, Serialize};
use serde_bare::{from_slice, to_vec};
// TODO: versioning V0
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
pub struct OverlayMeta {
pub users: u32,
pub last_used: Timestamp,
}
pub struct Overlay<'a> { use serde_bare::to_vec;
/// Overlay ID
id: OverlayId,
storage: &'a dyn KCVStorage,
}
impl<'a> Overlay<'a> { use crate::server_broker::OverlayInfo;
const PREFIX: u8 = b"o"[0]; use crate::server_broker::OverlayType;
// propertie's suffixes pub struct OverlayStorage<'a> {
const SECRET: u8 = b"s"[0]; key: Vec<u8>,
const PEER: u8 = b"p"[0]; overlay_type: ExistentialValue<OverlayType>,
const TOPIC: u8 = b"t"[0];
const META: u8 = b"m"[0];
const REPO: u8 = b"r"[0];
const ALL_PROPERTIES: [u8; 5] = [
Self::SECRET,
Self::PEER,
Self::TOPIC,
Self::META,
Self::REPO,
];
const SUFFIX_FOR_EXIST_CHECK: u8 = Self::SECRET;
pub fn open(id: &OverlayId, storage: &'a dyn KCVStorage) -> Result<Overlay<'a>, StorageError> {
let opening = Overlay {
id: id.clone(),
storage,
};
if !opening.exists() {
return Err(StorageError::NotFound);
}
Ok(opening)
}
pub fn create(
id: &OverlayId,
secret: &SymKey,
repo: Option<PubKey>,
storage: &'a dyn KCVStorage, storage: &'a dyn KCVStorage,
) -> Result<Overlay<'a>, StorageError> {
let acc = Overlay {
id: id.clone(),
storage,
};
if acc.exists() {
return Err(StorageError::BackendError);
} }
storage.write_transaction(&mut |tx| {
tx.put( impl<'a> IModel for OverlayStorage<'a> {
Self::PREFIX, fn key(&self) -> &Vec<u8> {
&to_vec(&id)?, &self.key
Some(Self::SECRET),
&to_vec(&secret)?,
&None,
)?;
if repo.is_some() {
tx.put(
Self::PREFIX,
&to_vec(&id)?,
Some(Self::REPO),
&to_vec(&repo.unwrap())?,
&None,
)?;
}
let meta = OverlayMeta {
users: 1,
last_used: now_timestamp(),
};
tx.put(
Self::PREFIX,
&to_vec(&id)?,
Some(Self::META),
&to_vec(&meta)?,
&None,
)?;
Ok(())
})?;
Ok(acc)
} }
pub fn exists(&self) -> bool { fn storage(&self) -> &dyn KCVStorage {
self.storage self.storage
.get(
Self::PREFIX,
&to_vec(&self.id).unwrap(),
Some(Self::SUFFIX_FOR_EXIST_CHECK),
&None,
)
.is_ok()
}
pub fn id(&self) -> OverlayId {
self.id
} }
pub fn add_peer(&self, peer: &PeerId) -> Result<(), StorageError> { fn class(&self) -> &Class {
if !self.exists() { &Self::CLASS
return Err(StorageError::BackendError);
} }
self.storage.put( fn existential(&mut self) -> Option<&mut dyn IExistentialValue> {
Self::PREFIX, Some(&mut self.overlay_type)
&to_vec(&self.id)?,
Some(Self::PEER),
&to_vec(peer)?,
&None,
)
} }
pub fn remove_peer(&self, peer: &PeerId) -> Result<(), StorageError> { // fn name(&self) -> String {
self.storage.del_property_value( // format_type_of(self)
Self::PREFIX, // }
&to_vec(&self.id)?,
Some(Self::PEER),
&to_vec(peer)?,
&None,
)
} }
pub fn has_peer(&self, peer: &PeerId) -> Result<(), StorageError> { impl<'a> OverlayStorage<'a> {
self.storage.has_property_value( const PREFIX: u8 = b'o';
Self::PREFIX,
&to_vec(&self.id)?,
Some(Self::PEER),
&to_vec(peer)?,
&None,
)
}
pub fn add_topic(&self, topic: &TopicId) -> Result<(), StorageError> { // Overlay properties
if !self.exists() { pub const TYPE: ExistentialValueColumn = ExistentialValueColumn::new(b'y');
return Err(StorageError::BackendError); pub const TOPIC: SingleValueColumn<Self, TopicId> = SingleValueColumn::new(b't');
}
self.storage.put(
Self::PREFIX,
&to_vec(&self.id)?,
Some(Self::TOPIC),
&to_vec(topic)?,
&None,
)
}
pub fn remove_topic(&self, topic: &TopicId) -> Result<(), StorageError> {
self.storage.del_property_value(
Self::PREFIX,
&to_vec(&self.id)?,
Some(Self::TOPIC),
&to_vec(topic)?,
&None,
)
}
pub fn has_topic(&self, topic: &TopicId) -> Result<(), StorageError> { // Overlay <-> Block refcount
self.storage.has_property_value( pub const BLOCKS: MultiCounterColumn<Self, BlockId> = MultiCounterColumn::new(b'b');
Self::PREFIX, // Overlay <-> Object refcount
&to_vec(&self.id)?, pub const OBJECTS: MultiCounterColumn<Self, ObjectId> = MultiCounterColumn::new(b'j');
Some(Self::TOPIC),
&to_vec(topic)?,
&None,
)
}
pub fn secret(&self) -> Result<SymKey, StorageError> { pub const CLASS: Class<'a> = Class::new(
match self "Overlay",
.storage Some(Self::PREFIX),
.get(Self::PREFIX, &to_vec(&self.id)?, Some(Self::SECRET), &None) Some(&Self::TYPE),
{ &[&Self::TOPIC as &dyn ISingleValueColumn],
Ok(secret) => Ok(from_slice::<SymKey>(&secret)?), &[&Self::BLOCKS as &dyn IMultiValueColumn, &Self::OBJECTS],
Err(e) => Err(e), );
}
}
pub fn metadata(&self) -> Result<OverlayMeta, StorageError> { pub fn new(id: &OverlayId, storage: &'a dyn KCVStorage) -> Self {
match self OverlayStorage {
.storage key: to_vec(id).unwrap(),
.get(Self::PREFIX, &to_vec(&self.id)?, Some(Self::META), &None) overlay_type: ExistentialValue::<OverlayType>::new(),
{ storage,
Ok(meta) => Ok(from_slice::<OverlayMeta>(&meta)?),
Err(e) => Err(e),
} }
} }
pub fn set_metadata(&self, meta: &OverlayMeta) -> Result<(), StorageError> {
if !self.exists() { pub fn load(id: &OverlayId, storage: &'a dyn KCVStorage) -> Result<OverlayInfo, StorageError> {
return Err(StorageError::BackendError); let mut opening = OverlayStorage::new(id, storage);
let props = opening.load_props()?;
let existential = col(&Self::TYPE, &props)?;
opening.overlay_type.set(&existential)?;
let loading = OverlayInfo {
overlay_type: existential,
overlay_topic: col(&Self::TOPIC, &props).ok(),
topics: HashMap::new(),
repos: HashMap::new(),
};
Ok(loading)
} }
self.storage.replace(
Self::PREFIX, pub fn open(
&to_vec(&self.id)?, id: &OverlayId,
Some(Self::META), storage: &'a dyn KCVStorage,
&to_vec(meta)?, ) -> Result<OverlayStorage<'a>, StorageError> {
&None, let mut opening = OverlayStorage::new(id, storage);
) opening.check_exists()?;
Ok(opening)
} }
pub fn repo(&self) -> Result<PubKey, StorageError> { pub fn create(
match self id: &OverlayId,
.storage overlay_type: &OverlayType,
.get(Self::PREFIX, &to_vec(&self.id)?, Some(Self::REPO), &None) storage: &'a mut dyn KCVStorage,
{ ) -> Result<OverlayStorage<'a>, StorageError> {
Ok(repo) => Ok(from_slice::<PubKey>(&repo)?), let mut overlay = OverlayStorage::new(id, storage);
Err(e) => Err(e), if overlay.exists() {
return Err(StorageError::AlreadyExists);
} }
overlay.overlay_type.set(overlay_type)?;
ExistentialValue::save(&overlay, overlay_type)?;
Ok(overlay)
} }
pub fn del(&self) -> Result<(), StorageError> { pub fn overlay_type(&mut self) -> &OverlayType {
self.storage.del_all( self.overlay_type.get().unwrap()
Self::PREFIX,
&to_vec(&self.id)?,
&Self::ALL_PROPERTIES,
&None,
)
} }
} }

@ -7,7 +7,7 @@
// notice may not be copied, modified, or distributed except // notice may not be copied, modified, or distributed except
// according to those terms. // according to those terms.
//! Repo OKM (Object Key/Col/Value Mapping) //! Repo Storage (Object Key/Col/Value Mapping)
use std::collections::HashMap; use std::collections::HashMap;
use std::collections::HashSet; use std::collections::HashSet;
@ -21,12 +21,12 @@ use serde_bare::to_vec;
use crate::server_broker::RepoInfo; use crate::server_broker::RepoInfo;
pub struct RepoOKM<'a> { pub struct RepoHashStorage<'a> {
key: Vec<u8>, key: Vec<u8>,
storage: &'a dyn KCVStorage, storage: &'a dyn KCVStorage,
} }
impl<'a> IModel for RepoOKM<'a> { impl<'a> IModel for RepoHashStorage<'a> {
fn key(&self) -> &Vec<u8> { fn key(&self) -> &Vec<u8> {
&self.key &self.key
} }
@ -41,7 +41,7 @@ impl<'a> IModel for RepoOKM<'a> {
} }
} }
impl<'a> RepoOKM<'a> { impl<'a> RepoHashStorage<'a> {
// RepoHash <-> Topic : list of topics of a repo that was pinned on the broker // RepoHash <-> Topic : list of topics of a repo that was pinned on the broker
pub const TOPICS: MultiValueColumn<Self, TopicId> = MultiValueColumn::new(b'r'); pub const TOPICS: MultiValueColumn<Self, TopicId> = MultiValueColumn::new(b'r');
// RepoHash <-> User : list of users who asked to expose the repo to the outer overlay // RepoHash <-> User : list of users who asked to expose the repo to the outer overlay
@ -80,7 +80,7 @@ impl<'a> RepoOKM<'a> {
repo: &RepoHash, repo: &RepoHash,
overlay: &OverlayId, overlay: &OverlayId,
storage: &'a dyn KCVStorage, storage: &'a dyn KCVStorage,
) -> Result<RepoOKM<'a>, StorageError> { ) -> Result<RepoHashStorage<'a>, StorageError> {
let mut opening = Self::new(repo, overlay, storage); let mut opening = Self::new(repo, overlay, storage);
Ok(opening) Ok(opening)
} }
@ -88,7 +88,7 @@ impl<'a> RepoOKM<'a> {
repo: &RepoHash, repo: &RepoHash,
overlay: &OverlayId, overlay: &OverlayId,
storage: &'a mut dyn KCVStorage, storage: &'a mut dyn KCVStorage,
) -> Result<RepoOKM<'a>, StorageError> { ) -> Result<RepoHashStorage<'a>, StorageError> {
let mut creating = Self::new(repo, overlay, storage); let mut creating = Self::new(repo, overlay, storage);
Ok(creating) Ok(creating)
} }

@ -7,7 +7,7 @@
// notice may not be copied, modified, or distributed except // notice may not be copied, modified, or distributed except
// according to those terms. // according to those terms.
//! Topic OKM (Object Key/Col/Value Mapping) //! Topic Storage (Object Key/Col/Value Mapping)
use std::collections::HashMap; use std::collections::HashMap;
use std::collections::HashSet; use std::collections::HashSet;
@ -21,13 +21,13 @@ use serde_bare::to_vec;
use crate::server_broker::TopicInfo; use crate::server_broker::TopicInfo;
pub struct Topic<'a> { pub struct TopicStorage<'a> {
key: Vec<u8>, key: Vec<u8>,
repo: ExistentialValue<RepoHash>, repo: ExistentialValue<RepoHash>,
storage: &'a dyn KCVStorage, storage: &'a dyn KCVStorage,
} }
impl<'a> IModel for Topic<'a> { impl<'a> IModel for TopicStorage<'a> {
fn key(&self) -> &Vec<u8> { fn key(&self) -> &Vec<u8> {
&self.key &self.key
} }
@ -45,7 +45,7 @@ impl<'a> IModel for Topic<'a> {
// } // }
} }
impl<'a> Topic<'a> { impl<'a> TopicStorage<'a> {
const PREFIX: u8 = b't'; const PREFIX: u8 = b't';
// Topic properties // Topic properties
@ -66,16 +66,26 @@ impl<'a> Topic<'a> {
&[&Self::USERS as &dyn IMultiValueColumn, &Self::HEADS], &[&Self::USERS as &dyn IMultiValueColumn, &Self::HEADS],
); );
pub fn new(id: &TopicId, overlay: &OverlayId, storage: &'a dyn KCVStorage) -> Self {
let mut key: Vec<u8> = Vec::with_capacity(33 + 33);
key.append(&mut to_vec(overlay).unwrap());
key.append(&mut to_vec(id).unwrap());
TopicStorage {
key,
repo: ExistentialValue::<RepoHash>::new(),
storage,
}
}
pub fn load( pub fn load(
id: &TopicId, id: &TopicId,
overlay: &OverlayId, overlay: &OverlayId,
storage: &'a dyn KCVStorage, storage: &'a dyn KCVStorage,
) -> Result<TopicInfo, StorageError> { ) -> Result<TopicInfo, StorageError> {
let mut opening = Topic::new(id, overlay, storage); let mut opening = TopicStorage::new(id, overlay, storage);
let props = opening.load_props()?; let props = opening.load_props()?;
let existential = col(&Self::REPO, &props)?; let existential = col(&Self::REPO, &props)?;
opening.repo.set(&existential)?; opening.repo.set(&existential)?;
//ExistentialValue::save(&opening, &existential)?;
let ti = TopicInfo { let ti = TopicInfo {
repo: existential, repo: existential,
publisher_advert: col(&Self::ADVERT, &props).ok(), publisher_advert: col(&Self::ADVERT, &props).ok(),
@ -86,23 +96,12 @@ impl<'a> Topic<'a> {
Ok(ti) Ok(ti)
} }
pub fn new(id: &TopicId, overlay: &OverlayId, storage: &'a dyn KCVStorage) -> Self {
let mut key: Vec<u8> = Vec::with_capacity(33 + 33);
key.append(&mut to_vec(overlay).unwrap());
key.append(&mut to_vec(id).unwrap());
Topic {
key,
repo: ExistentialValue::<RepoHash>::new(),
storage,
}
}
pub fn open( pub fn open(
id: &TopicId, id: &TopicId,
overlay: &OverlayId, overlay: &OverlayId,
storage: &'a dyn KCVStorage, storage: &'a dyn KCVStorage,
) -> Result<Topic<'a>, StorageError> { ) -> Result<TopicStorage<'a>, StorageError> {
let mut opening = Topic::new(id, overlay, storage); let mut opening = TopicStorage::new(id, overlay, storage);
opening.check_exists()?; opening.check_exists()?;
Ok(opening) Ok(opening)
} }
@ -111,8 +110,8 @@ impl<'a> Topic<'a> {
overlay: &OverlayId, overlay: &OverlayId,
repo: &RepoHash, repo: &RepoHash,
storage: &'a mut dyn KCVStorage, storage: &'a mut dyn KCVStorage,
) -> Result<Topic<'a>, StorageError> { ) -> Result<TopicStorage<'a>, StorageError> {
let mut topic = Topic::new(id, overlay, storage); let mut topic = TopicStorage::new(id, overlay, storage);
if topic.exists() { if topic.exists() {
return Err(StorageError::AlreadyExists); return Err(StorageError::AlreadyExists);
} }

@ -264,6 +264,7 @@ impl<
Ok(res) Ok(res)
} }
} }
impl< impl<
Model: IModel, Model: IModel,
Column: Eq + PartialEq + Hash + Serialize + Default + for<'d> Deserialize<'d>, Column: Eq + PartialEq + Hash + Serialize + Default + for<'d> Deserialize<'d>,
@ -397,6 +398,103 @@ impl<
} }
} }
pub struct MultiCounterColumn<
Model: IModel,
Column: Eq + PartialEq + Hash + Serialize + Default + for<'a> Deserialize<'a>,
> {
prefix: u8,
phantom_column: PhantomData<Column>,
phantom_model: PhantomData<Model>,
}
impl<
Model: IModel,
Column: Eq + PartialEq + Hash + Serialize + Default + for<'d> Deserialize<'d>,
> MultiCounterColumn<Model, Column>
{
pub const fn new(prefix: u8) -> Self {
MultiCounterColumn {
prefix,
phantom_column: PhantomData,
phantom_model: PhantomData,
}
}
pub fn increment(&self, model: &mut Model, column: &Column) -> Result<(), StorageError> {
let key = MultiValueColumn::compute_key(model, column)?;
model.storage().write_transaction(&mut |tx| {
let mut val: u64 = match tx.get(self.prefix, &key, None, &None) {
Ok(val_ser) => from_slice(&val_ser)?,
Err(StorageError::NotFound) => 0,
Err(e) => return Err(e),
};
val += 1;
let val_ser = to_vec(&val)?;
tx.put(self.prefix, &key, None, &val_ser, &None)?;
Ok(())
})
}
/// returns true if the counter reached zero (and the key was removed from KVC store)
pub fn decrement(&self, model: &mut Model, column: &Column) -> Result<bool, StorageError> {
let key = MultiValueColumn::compute_key(model, column)?;
let mut ret: bool = false;
model.storage().write_transaction(&mut |tx| {
let val_ser = tx.get(self.prefix, &key, None, &None)?;
let mut val: u64 = from_slice(&val_ser)?;
val -= 1;
ret = val == 0;
if ret {
tx.del(self.prefix, &key, None, &None)?;
} else {
let val_ser = to_vec(&val)?;
tx.put(self.prefix, &key, None, &val_ser, &None)?;
}
Ok(())
})?;
Ok(ret)
}
pub fn get(&self, model: &mut Model, column: &Column) -> Result<u64, StorageError> {
let key = MultiValueColumn::compute_key(model, column)?;
let val_ser = model.storage().get(self.prefix, &key, None, &None)?;
let val: u64 = from_slice(&val_ser)?;
Ok(val)
}
pub fn get_all(&self, model: &mut Model) -> Result<HashMap<Column, u64>, StorageError> {
model.check_exists()?;
let key_prefix = model.key();
let key_prefix_len = key_prefix.len();
let mut res: HashMap<Column, u64> = HashMap::new();
let total_size = key_prefix_len + self.value_size()?;
for val in model.storage().get_all_keys_and_values(
self.prefix,
total_size,
key_prefix.to_vec(),
None,
&None,
)? {
if val.0.len() == total_size + 1 {
let col: Column = from_slice(&val.0[1 + key_prefix_len..total_size + 1])?;
let val = from_slice(&val.1)?;
res.insert(col, val);
}
}
Ok(res)
}
}
impl<
Model: IModel,
Column: Eq + PartialEq + Hash + Serialize + Default + for<'d> Deserialize<'d>,
> IMultiValueColumn for MultiCounterColumn<Model, Column>
{
fn value_size(&self) -> Result<usize, StorageError> {
Ok(to_vec(&(0 as u64))?.len())
}
fn prefix(&self) -> u8 {
self.prefix
}
}
pub trait ISingleValueColumn { pub trait ISingleValueColumn {
fn suffix(&self) -> u8; fn suffix(&self) -> u8;
} }

@ -7,7 +7,7 @@
// notice may not be copied, modified, or distributed except // notice may not be copied, modified, or distributed except
// according to those terms. // according to those terms.
//! Branch storage OKM (Object Key/Col/Value Mapping) //! Branch Storage (Object Key/Col/Value Mapping)
use std::collections::hash_map::DefaultHasher; use std::collections::hash_map::DefaultHasher;
use std::collections::HashMap; use std::collections::HashMap;

@ -7,7 +7,7 @@
// notice may not be copied, modified, or distributed except // notice may not be copied, modified, or distributed except
// according to those terms. // according to those terms.
//! Repo storage OKM (Object Key/Col/Value Mapping) //! Repo Storage (Object Key/Col/Value Mapping)
use std::collections::hash_map::DefaultHasher; use std::collections::hash_map::DefaultHasher;
use std::collections::HashMap; use std::collections::HashMap;

Loading…
Cancel
Save