BROKEN: start work on hash discriminant

This commit is contained in:
asonix 2023-08-13 22:06:42 -05:00
parent e4e93cddf9
commit 3129f7844e
5 changed files with 220 additions and 92 deletions

View file

@ -7,7 +7,9 @@ use std::str::FromStr;
pub(crate) use animation::{AnimationFormat, AnimationOutput}; pub(crate) use animation::{AnimationFormat, AnimationOutput};
pub(crate) use image::{ImageFormat, ImageInput, ImageOutput}; pub(crate) use image::{ImageFormat, ImageInput, ImageOutput};
pub(crate) use video::{InternalVideoFormat, OutputVideoFormat, VideoFormat, VideoCodec, AudioCodec}; pub(crate) use video::{
AudioCodec, InternalVideoFormat, OutputVideoFormat, VideoCodec, VideoFormat,
};
#[derive(Clone, Debug)] #[derive(Clone, Debug)]
pub(crate) struct Validations<'a> { pub(crate) struct Validations<'a> {
@ -75,6 +77,39 @@ impl InternalFormat {
} }
} }
pub(crate) const fn to_bytes(self) -> &'static [u8] {
match self {
Self::Animation(AnimationFormat::Apng) => b"a-apng",
Self::Animation(AnimationFormat::Avif) => b"a-avif",
Self::Animation(AnimationFormat::Gif) => b"a-gif",
Self::Animation(AnimationFormat::Webp) => b"a-webp",
Self::Image(ImageFormat::Avif) => b"i-avif",
Self::Image(ImageFormat::Jpeg) => b"i-jpeg",
Self::Image(ImageFormat::Jxl) => b"i-jxl",
Self::Image(ImageFormat::Png) => b"i-png",
Self::Image(ImageFormat::Webp) => b"i-webp",
Self::Video(InternalVideoFormat::Mp4) => b"v-mp4",
Self::Video(InternalVideoFormat::Webm) => b"v-webm",
}
}
pub(crate) const fn from_bytes(bytes: &[u8]) -> Option<Self> {
match bytes {
b"a-apng" => Some(Self::Animation(AnimationFormat::Apng)),
b"a-avif" => Some(Self::Animation(AnimationFormat::Avif)),
b"a-gif" => Some(Self::Animation(AnimationFormat::Gif)),
b"a-webp" => Some(Self::Animation(AnimationFormat::Webp)),
b"i-avif" => Some(Self::Image(ImageFormat::Avif)),
b"i-jpeg" => Some(Self::Image(ImageFormat::Jpeg)),
b"i-jxl" => Some(Self::Image(ImageFormat::Jxl)),
b"i-png" => Some(Self::Image(ImageFormat::Png)),
b"i-webp" => Some(Self::Image(ImageFormat::Webp)),
b"v-mp4" => Some(Self::Video(InternalVideoFormat::Mp4)),
b"v-webm" => Some(Self::Video(InternalVideoFormat::Webm)),
_ => None,
}
}
pub(crate) fn maybe_from_media_type(mime: &mime::Mime, has_frames: bool) -> Option<Self> { pub(crate) fn maybe_from_media_type(mime: &mime::Mime, has_frames: bool) -> Option<Self> {
match (mime.type_(), mime.subtype().as_str(), has_frames) { match (mime.type_(), mime.subtype().as_str(), has_frames) {
(mime::IMAGE, "apng", _) => Some(Self::Animation(AnimationFormat::Apng)), (mime::IMAGE, "apng", _) => Some(Self::Animation(AnimationFormat::Apng)),

View file

@ -7,12 +7,17 @@ use std::{
}; };
use tokio::io::{AsyncRead, ReadBuf}; use tokio::io::{AsyncRead, ReadBuf};
struct State<D> {
hasher: D,
size: u64,
}
pin_project_lite::pin_project! { pin_project_lite::pin_project! {
pub(crate) struct Hasher<I, D> { pub(crate) struct Hasher<I, D> {
#[pin] #[pin]
inner: I, inner: I,
hasher: Rc<RefCell<D>>, state: Rc<RefCell<State<D>>>,
} }
} }
@ -23,12 +28,15 @@ where
pub(super) fn new(reader: I, digest: D) -> Self { pub(super) fn new(reader: I, digest: D) -> Self {
Hasher { Hasher {
inner: reader, inner: reader,
hasher: Rc::new(RefCell::new(digest)), state: Rc::new(RefCell::new(State {
hasher: digest,
size: 0,
})),
} }
} }
pub(super) fn hasher(&self) -> Rc<RefCell<D>> { pub(super) fn state(&self) -> Rc<RefCell<State<D>>> {
Rc::clone(&self.hasher) Rc::clone(&self.state)
} }
} }
@ -45,15 +53,15 @@ where
let this = self.as_mut().project(); let this = self.as_mut().project();
let reader = this.inner; let reader = this.inner;
let hasher = this.hasher; let state = this.state;
let before_len = buf.filled().len(); let before_len = buf.filled().len();
let poll_res = reader.poll_read(cx, buf); let poll_res = reader.poll_read(cx, buf);
let after_len = buf.filled().len(); let after_len = buf.filled().len();
if after_len > before_len { if after_len > before_len {
hasher let mut guard = state.borrow_mut();
.borrow_mut() guard.hasher.update(&buf.filled()[before_len..after_len]);
.update(&buf.filled()[before_len..after_len]); guard.size += u64::try_from(after_len - before_len).expect("Size is reasonable");
} }
poll_res poll_res
} }
@ -93,7 +101,7 @@ mod test {
tokio::io::copy(&mut reader, &mut tokio::io::sink()).await?; tokio::io::copy(&mut reader, &mut tokio::io::sink()).await?;
Ok(reader.hasher().borrow_mut().finalize_reset().to_vec()) as std::io::Result<_> Ok(reader.state().borrow_mut().hasher.finalize_reset().to_vec()) as std::io::Result<_>
}) })
.unwrap(); .unwrap();

View file

@ -8,8 +8,11 @@ use std::fmt::Debug;
use url::Url; use url::Url;
use uuid::Uuid; use uuid::Uuid;
mod hash;
pub(crate) mod sled; pub(crate) mod sled;
pub(crate) use hash::Hash;
#[derive(Clone, Debug)] #[derive(Clone, Debug)]
pub(crate) enum Repo { pub(crate) enum Repo {
Sled(self::sled::SledRepo), Sled(self::sled::SledRepo),
@ -207,17 +210,16 @@ where
pub(crate) trait VariantAccessRepo: BaseRepo { pub(crate) trait VariantAccessRepo: BaseRepo {
type VariantAccessStream: Stream<Item = Result<(Self::Bytes, String), RepoError>>; type VariantAccessStream: Stream<Item = Result<(Self::Bytes, String), RepoError>>;
async fn accessed(&self, hash: Self::Bytes, variant: String) -> Result<(), RepoError>; async fn accessed(&self, hash: Hash, variant: String) -> Result<(), RepoError>;
async fn contains_variant(&self, hash: Self::Bytes, variant: String) async fn contains_variant(&self, hash: Hash, variant: String) -> Result<bool, RepoError>;
-> Result<bool, RepoError>;
async fn older_variants( async fn older_variants(
&self, &self,
timestamp: time::OffsetDateTime, timestamp: time::OffsetDateTime,
) -> Result<Self::VariantAccessStream, RepoError>; ) -> Result<Self::VariantAccessStream, RepoError>;
async fn remove_access(&self, hash: Self::Bytes, variant: String) -> Result<(), RepoError>; async fn remove_access(&self, hash: Hash, variant: String) -> Result<(), RepoError>;
} }
#[async_trait::async_trait(?Send)] #[async_trait::async_trait(?Send)]
@ -227,15 +229,11 @@ where
{ {
type VariantAccessStream = T::VariantAccessStream; type VariantAccessStream = T::VariantAccessStream;
async fn accessed(&self, hash: Self::Bytes, variant: String) -> Result<(), RepoError> { async fn accessed(&self, hash: Hash, variant: String) -> Result<(), RepoError> {
T::accessed(self, hash, variant).await T::accessed(self, hash, variant).await
} }
async fn contains_variant( async fn contains_variant(&self, hash: Hash, variant: String) -> Result<bool, RepoError> {
&self,
hash: Self::Bytes,
variant: String,
) -> Result<bool, RepoError> {
T::contains_variant(self, hash, variant).await T::contains_variant(self, hash, variant).await
} }
@ -246,7 +244,7 @@ where
T::older_variants(self, timestamp).await T::older_variants(self, timestamp).await
} }
async fn remove_access(&self, hash: Self::Bytes, variant: String) -> Result<(), RepoError> { async fn remove_access(&self, hash: Hash, variant: String) -> Result<(), RepoError> {
T::remove_access(self, hash, variant).await T::remove_access(self, hash, variant).await
} }
} }
@ -436,7 +434,7 @@ where
#[async_trait::async_trait(?Send)] #[async_trait::async_trait(?Send)]
pub(crate) trait HashRepo: BaseRepo { pub(crate) trait HashRepo: BaseRepo {
type Stream: Stream<Item = Result<Self::Bytes, RepoError>>; type Stream: Stream<Item = Result<Hash, RepoError>>;
async fn size(&self) -> Result<u64, RepoError>; async fn size(&self) -> Result<u64, RepoError>;
@ -444,49 +442,49 @@ pub(crate) trait HashRepo: BaseRepo {
async fn create<I: Identifier>( async fn create<I: Identifier>(
&self, &self,
hash: Self::Bytes, hash: Hash,
identifier: &I, identifier: &I,
) -> Result<Result<(), HashAlreadyExists>, StoreError>; ) -> Result<Result<(), HashAlreadyExists>, StoreError>;
async fn update_identifier<I: Identifier>( async fn update_identifier<I: Identifier>(
&self, &self,
hash: Self::Bytes, hash: Hash,
identifier: &I, identifier: &I,
) -> Result<(), StoreError>; ) -> Result<(), StoreError>;
async fn identifier<I: Identifier + 'static>( async fn identifier<I: Identifier + 'static>(
&self, &self,
hash: Self::Bytes, hash: Hash,
) -> Result<Option<I>, StoreError>; ) -> Result<Option<I>, StoreError>;
async fn relate_variant_identifier<I: Identifier>( async fn relate_variant_identifier<I: Identifier>(
&self, &self,
hash: Self::Bytes, hash: Hash,
variant: String, variant: String,
identifier: &I, identifier: &I,
) -> Result<(), StoreError>; ) -> Result<(), StoreError>;
async fn variant_identifier<I: Identifier + 'static>( async fn variant_identifier<I: Identifier + 'static>(
&self, &self,
hash: Self::Bytes, hash: Hash,
variant: String, variant: String,
) -> Result<Option<I>, StoreError>; ) -> Result<Option<I>, StoreError>;
async fn variants<I: Identifier + 'static>( async fn variants<I: Identifier + 'static>(
&self, &self,
hash: Self::Bytes, hash: Hash,
) -> Result<Vec<(String, I)>, StoreError>; ) -> Result<Vec<(String, I)>, StoreError>;
async fn remove_variant(&self, hash: Self::Bytes, variant: String) -> Result<(), RepoError>; async fn remove_variant(&self, hash: Hash, variant: String) -> Result<(), RepoError>;
async fn relate_motion_identifier<I: Identifier>( async fn relate_motion_identifier<I: Identifier>(
&self, &self,
hash: Self::Bytes, hash: Hash,
identifier: &I, identifier: &I,
) -> Result<(), StoreError>; ) -> Result<(), StoreError>;
async fn motion_identifier<I: Identifier + 'static>( async fn motion_identifier<I: Identifier + 'static>(
&self, &self,
hash: Self::Bytes, hash: Hash,
) -> Result<Option<I>, StoreError>; ) -> Result<Option<I>, StoreError>;
async fn cleanup(&self, hash: Self::Bytes) -> Result<(), RepoError>; async fn cleanup(&self, hash: Hash) -> Result<(), RepoError>;
} }
#[async_trait::async_trait(?Send)] #[async_trait::async_trait(?Send)]
@ -506,7 +504,7 @@ where
async fn create<I: Identifier>( async fn create<I: Identifier>(
&self, &self,
hash: Self::Bytes, hash: Hash,
identifier: &I, identifier: &I,
) -> Result<Result<(), HashAlreadyExists>, StoreError> { ) -> Result<Result<(), HashAlreadyExists>, StoreError> {
T::create(self, hash, identifier).await T::create(self, hash, identifier).await
@ -514,7 +512,7 @@ where
async fn update_identifier<I: Identifier>( async fn update_identifier<I: Identifier>(
&self, &self,
hash: Self::Bytes, hash: Hash,
identifier: &I, identifier: &I,
) -> Result<(), StoreError> { ) -> Result<(), StoreError> {
T::update_identifier(self, hash, identifier).await T::update_identifier(self, hash, identifier).await
@ -522,14 +520,14 @@ where
async fn identifier<I: Identifier + 'static>( async fn identifier<I: Identifier + 'static>(
&self, &self,
hash: Self::Bytes, hash: Hash,
) -> Result<Option<I>, StoreError> { ) -> Result<Option<I>, StoreError> {
T::identifier(self, hash).await T::identifier(self, hash).await
} }
async fn relate_variant_identifier<I: Identifier>( async fn relate_variant_identifier<I: Identifier>(
&self, &self,
hash: Self::Bytes, hash: Hash,
variant: String, variant: String,
identifier: &I, identifier: &I,
) -> Result<(), StoreError> { ) -> Result<(), StoreError> {
@ -538,7 +536,7 @@ where
async fn variant_identifier<I: Identifier + 'static>( async fn variant_identifier<I: Identifier + 'static>(
&self, &self,
hash: Self::Bytes, hash: Hash,
variant: String, variant: String,
) -> Result<Option<I>, StoreError> { ) -> Result<Option<I>, StoreError> {
T::variant_identifier(self, hash, variant).await T::variant_identifier(self, hash, variant).await
@ -546,18 +544,18 @@ where
async fn variants<I: Identifier + 'static>( async fn variants<I: Identifier + 'static>(
&self, &self,
hash: Self::Bytes, hash: Hash,
) -> Result<Vec<(String, I)>, StoreError> { ) -> Result<Vec<(String, I)>, StoreError> {
T::variants(self, hash).await T::variants(self, hash).await
} }
async fn remove_variant(&self, hash: Self::Bytes, variant: String) -> Result<(), RepoError> { async fn remove_variant(&self, hash: Hash, variant: String) -> Result<(), RepoError> {
T::remove_variant(self, hash, variant).await T::remove_variant(self, hash, variant).await
} }
async fn relate_motion_identifier<I: Identifier>( async fn relate_motion_identifier<I: Identifier>(
&self, &self,
hash: Self::Bytes, hash: Hash,
identifier: &I, identifier: &I,
) -> Result<(), StoreError> { ) -> Result<(), StoreError> {
T::relate_motion_identifier(self, hash, identifier).await T::relate_motion_identifier(self, hash, identifier).await
@ -565,12 +563,12 @@ where
async fn motion_identifier<I: Identifier + 'static>( async fn motion_identifier<I: Identifier + 'static>(
&self, &self,
hash: Self::Bytes, hash: Hash,
) -> Result<Option<I>, StoreError> { ) -> Result<Option<I>, StoreError> {
T::motion_identifier(self, hash).await T::motion_identifier(self, hash).await
} }
async fn cleanup(&self, hash: Self::Bytes) -> Result<(), RepoError> { async fn cleanup(&self, hash: Hash) -> Result<(), RepoError> {
T::cleanup(self, hash).await T::cleanup(self, hash).await
} }
} }
@ -581,14 +579,14 @@ pub(crate) trait AliasRepo: BaseRepo {
&self, &self,
alias: &Alias, alias: &Alias,
delete_token: &DeleteToken, delete_token: &DeleteToken,
hash: Self::Bytes, hash: Hash,
) -> Result<Result<(), AliasAlreadyExists>, RepoError>; ) -> Result<Result<(), AliasAlreadyExists>, RepoError>;
async fn delete_token(&self, alias: &Alias) -> Result<Option<DeleteToken>, RepoError>; async fn delete_token(&self, alias: &Alias) -> Result<Option<DeleteToken>, RepoError>;
async fn hash(&self, alias: &Alias) -> Result<Option<Self::Bytes>, RepoError>; async fn hash(&self, alias: &Alias) -> Result<Option<Hash>, RepoError>;
async fn for_hash(&self, hash: Self::Bytes) -> Result<Vec<Alias>, RepoError>; async fn for_hash(&self, hash: Hash) -> Result<Vec<Alias>, RepoError>;
async fn cleanup(&self, alias: &Alias) -> Result<(), RepoError>; async fn cleanup(&self, alias: &Alias) -> Result<(), RepoError>;
} }
@ -602,7 +600,7 @@ where
&self, &self,
alias: &Alias, alias: &Alias,
delete_token: &DeleteToken, delete_token: &DeleteToken,
hash: Self::Bytes, hash: Hash,
) -> Result<Result<(), AliasAlreadyExists>, RepoError> { ) -> Result<Result<(), AliasAlreadyExists>, RepoError> {
T::create(self, alias, delete_token, hash).await T::create(self, alias, delete_token, hash).await
} }
@ -611,11 +609,11 @@ where
T::delete_token(self, alias).await T::delete_token(self, alias).await
} }
async fn hash(&self, alias: &Alias) -> Result<Option<Self::Bytes>, RepoError> { async fn hash(&self, alias: &Alias) -> Result<Option<Hash>, RepoError> {
T::hash(self, alias).await T::hash(self, alias).await
} }
async fn for_hash(&self, hash: Self::Bytes) -> Result<Vec<Alias>, RepoError> { async fn for_hash(&self, hash: Hash) -> Result<Vec<Alias>, RepoError> {
T::for_hash(self, hash).await T::for_hash(self, hash).await
} }

65
src/repo/hash.rs Normal file
View file

@ -0,0 +1,65 @@
use crate::formats::InternalFormat;
use std::sync::Arc;
#[derive(Clone)]
pub(crate) struct Hash {
hash: Arc<[u8; 32]>,
size: u64,
format: InternalFormat,
}
impl Hash {
pub(crate) fn new(hash: Arc<[u8; 32]>, size: u64, format: InternalFormat) -> Self {
Self { hash, format, size }
}
pub(super) fn to_bytes(&self) -> Vec<u8> {
let format = self.format.to_bytes();
let mut vec = Vec::with_capacity(32 + 8 + format.len());
vec.extend_from_slice(&self.hash[..]);
vec.extend(self.size.to_be_bytes());
vec.extend(format);
vec
}
pub(super) fn to_ivec(&self) -> sled::IVec {
sled::IVec::from(self.to_bytes())
}
pub(super) fn from_ivec(ivec: sled::IVec) -> Option<Self> {
Self::from_bytes(&ivec)
}
pub(super) fn from_bytes(bytes: &[u8]) -> Option<Self> {
if bytes.len() < 32 + 8 + 5 {
return None;
}
let hash = &bytes[..32];
let size = &bytes[32..40];
let format = &bytes[40..];
let hash: [u8; 32] = hash.try_into().expect("Correct length");
let size: [u8; 8] = size.try_into().expect("Correct length");
let format = InternalFormat::from_bytes(format)?;
Some(Self {
hash: Arc::new(hash),
size: u64::from_be_bytes(size),
format,
})
}
}
impl std::fmt::Debug for Hash {
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
f.debug_struct("Hash")
.field("hash", &hex::encode(&*self.hash))
.field("format", &self.format)
.field("size", &self.size)
.finish()
}
}

View file

@ -1,7 +1,7 @@
use crate::{ use crate::{
details::MaybeHumanDate, details::MaybeHumanDate,
repo::{ repo::{
Alias, AliasAlreadyExists, AliasRepo, BaseRepo, DeleteToken, Details, FullRepo, hash::Hash, Alias, AliasAlreadyExists, AliasRepo, BaseRepo, DeleteToken, Details, FullRepo,
HashAlreadyExists, HashRepo, Identifier, IdentifierRepo, JobId, MigrationRepo, QueueRepo, HashAlreadyExists, HashRepo, Identifier, IdentifierRepo, JobId, MigrationRepo, QueueRepo,
SettingsRepo, UploadId, UploadRepo, UploadResult, SettingsRepo, UploadId, UploadRepo, UploadResult,
}, },
@ -405,8 +405,9 @@ impl AliasAccessRepo for SledRepo {
impl VariantAccessRepo for SledRepo { impl VariantAccessRepo for SledRepo {
type VariantAccessStream = VariantAccessStream; type VariantAccessStream = VariantAccessStream;
#[tracing::instrument(level = "debug", skip_all, fields(hash = %hex::encode(&hash), variant = %variant))] #[tracing::instrument(level = "debug", skip(self))]
async fn accessed(&self, hash: Self::Bytes, variant: String) -> Result<(), RepoError> { async fn accessed(&self, hash: Hash, variant: String) -> Result<(), RepoError> {
let hash = hash.to_bytes();
let key = variant_access_key(&hash, &variant); let key = variant_access_key(&hash, &variant);
let now_string = time::OffsetDateTime::now_utc() let now_string = time::OffsetDateTime::now_utc()
@ -428,12 +429,9 @@ impl VariantAccessRepo for SledRepo {
.map_err(RepoError::from) .map_err(RepoError::from)
} }
#[tracing::instrument(level = "debug", skip_all, fields(hash = %hex::encode(&hash), variant = %variant))] #[tracing::instrument(level = "debug", skip(self))]
async fn contains_variant( async fn contains_variant(&self, hash: Hash, variant: String) -> Result<bool, RepoError> {
&self, let hash = hash.to_bytes();
hash: Self::Bytes,
variant: String,
) -> Result<bool, RepoError> {
let key = variant_access_key(&hash, &variant); let key = variant_access_key(&hash, &variant);
let timestamp = b!(self.variant_access, variant_access.get(key)); let timestamp = b!(self.variant_access, variant_access.get(key));
@ -465,8 +463,9 @@ impl VariantAccessRepo for SledRepo {
}) })
} }
#[tracing::instrument(level = "debug", skip_all, fields(hash = %hex::encode(&hash), variant = %variant))] #[tracing::instrument(level = "debug", skip(self))]
async fn remove_access(&self, hash: Self::Bytes, variant: String) -> Result<(), RepoError> { async fn remove_access(&self, hash: Hash, variant: String) -> Result<(), RepoError> {
let hash = hash.to_bytes();
let key = variant_access_key(&hash, &variant); let key = variant_access_key(&hash, &variant);
let variant_access = self.variant_access.clone(); let variant_access = self.variant_access.clone();
@ -1012,7 +1011,7 @@ impl MigrationRepo for SledRepo {
} }
} }
type StreamItem = Result<IVec, RepoError>; type StreamItem = Result<Hash, RepoError>;
type LocalBoxStream<'a, T> = Pin<Box<dyn Stream<Item = T> + 'a>>; type LocalBoxStream<'a, T> = Pin<Box<dyn Stream<Item = T> + 'a>>;
#[async_trait::async_trait(?Send)] #[async_trait::async_trait(?Send)]
@ -1028,19 +1027,20 @@ impl HashRepo for SledRepo {
} }
async fn hashes(&self) -> Self::Stream { async fn hashes(&self) -> Self::Stream {
let iter = self let iter = self.hashes.iter().keys().filter_map(|res| {
.hashes res.map_err(SledError::from)
.iter() .map_err(RepoError::from)
.keys() .map(Hash::from_ivec)
.map(|res| res.map_err(SledError::from).map_err(RepoError::from)); .transpose()
});
Box::pin(from_iterator(iter, 8)) Box::pin(from_iterator(iter, 8))
} }
#[tracing::instrument(level = "trace", skip(self, hash), fields(hash = hex::encode(&hash)))] #[tracing::instrument(level = "trace", skip(self))]
async fn create<I: Identifier>( async fn create<I: Identifier>(
&self, &self,
hash: Self::Bytes, hash: Hash,
identifier: &I, identifier: &I,
) -> Result<Result<(), HashAlreadyExists>, StoreError> { ) -> Result<Result<(), HashAlreadyExists>, StoreError> {
let identifier: sled::IVec = identifier.to_bytes()?.into(); let identifier: sled::IVec = identifier.to_bytes()?.into();
@ -1048,14 +1048,16 @@ impl HashRepo for SledRepo {
let hashes = self.hashes.clone(); let hashes = self.hashes.clone();
let hash_identifiers = self.hash_identifiers.clone(); let hash_identifiers = self.hash_identifiers.clone();
let hash = hash.to_ivec();
let res = actix_web::web::block(move || { let res = actix_web::web::block(move || {
(&hashes, &hash_identifiers).transaction(|(hashes, hash_identifiers)| { (&hashes, &hash_identifiers).transaction(|(hashes, hash_identifiers)| {
if hashes.get(&hash)?.is_some() { if hashes.get(hash.clone())?.is_some() {
return Ok(Err(HashAlreadyExists)); return Ok(Err(HashAlreadyExists));
} }
hashes.insert(&hash, &hash)?; hashes.insert(hash.clone(), hash.clone())?;
hash_identifiers.insert(&hash, &identifier)?; hash_identifiers.insert(hash.clone(), &identifier)?;
Ok(Ok(())) Ok(Ok(()))
}) })
@ -1073,11 +1075,13 @@ impl HashRepo for SledRepo {
async fn update_identifier<I: Identifier>( async fn update_identifier<I: Identifier>(
&self, &self,
hash: Self::Bytes, hash: Hash,
identifier: &I, identifier: &I,
) -> Result<(), StoreError> { ) -> Result<(), StoreError> {
let identifier = identifier.to_bytes()?; let identifier = identifier.to_bytes()?;
let hash = hash.to_ivec();
b!( b!(
self.hash_identifiers, self.hash_identifiers,
hash_identifiers.insert(hash, identifier) hash_identifiers.insert(hash, identifier)
@ -1086,11 +1090,13 @@ impl HashRepo for SledRepo {
Ok(()) Ok(())
} }
#[tracing::instrument(level = "trace", skip(self, hash), fields(hash = hex::encode(&hash)))] #[tracing::instrument(level = "trace", skip(self))]
async fn identifier<I: Identifier + 'static>( async fn identifier<I: Identifier + 'static>(
&self, &self,
hash: Self::Bytes, hash: Hash,
) -> Result<Option<I>, StoreError> { ) -> Result<Option<I>, StoreError> {
let hash = hash.to_ivec();
let Some(ivec) = b!(self.hash_identifiers, hash_identifiers.get(hash)) else { let Some(ivec) = b!(self.hash_identifiers, hash_identifiers.get(hash)) else {
return Ok(None); return Ok(None);
}; };
@ -1098,13 +1104,15 @@ impl HashRepo for SledRepo {
Ok(Some(I::from_bytes(ivec.to_vec())?)) Ok(Some(I::from_bytes(ivec.to_vec())?))
} }
#[tracing::instrument(level = "trace", skip(self, hash, identifier), fields(hash = hex::encode(&hash), identifier = identifier.string_repr()))] #[tracing::instrument(level = "trace", skip(self, identifier), fields(identifier = identifier.string_repr()))]
async fn relate_variant_identifier<I: Identifier>( async fn relate_variant_identifier<I: Identifier>(
&self, &self,
hash: Self::Bytes, hash: Hash,
variant: String, variant: String,
identifier: &I, identifier: &I,
) -> Result<(), StoreError> { ) -> Result<(), StoreError> {
let hash = hash.to_bytes();
let key = variant_key(&hash, &variant); let key = variant_key(&hash, &variant);
let value = identifier.to_bytes()?; let value = identifier.to_bytes()?;
@ -1116,12 +1124,14 @@ impl HashRepo for SledRepo {
Ok(()) Ok(())
} }
#[tracing::instrument(level = "trace", skip(self, hash), fields(hash = hex::encode(&hash)))] #[tracing::instrument(level = "trace", skip(self))]
async fn variant_identifier<I: Identifier + 'static>( async fn variant_identifier<I: Identifier + 'static>(
&self, &self,
hash: Self::Bytes, hash: Hash,
variant: String, variant: String,
) -> Result<Option<I>, StoreError> { ) -> Result<Option<I>, StoreError> {
let hash = hash.to_bytes();
let key = variant_key(&hash, &variant); let key = variant_key(&hash, &variant);
let opt = b!( let opt = b!(
@ -1132,15 +1142,17 @@ impl HashRepo for SledRepo {
opt.map(|ivec| I::from_bytes(ivec.to_vec())).transpose() opt.map(|ivec| I::from_bytes(ivec.to_vec())).transpose()
} }
#[tracing::instrument(level = "debug", skip(self, hash), fields(hash = hex::encode(&hash)))] #[tracing::instrument(level = "debug", skip(self))]
async fn variants<I: Identifier + 'static>( async fn variants<I: Identifier + 'static>(
&self, &self,
hash: Self::Bytes, hash: Hash,
) -> Result<Vec<(String, I)>, StoreError> { ) -> Result<Vec<(String, I)>, StoreError> {
let hash = hash.to_ivec();
let vec = b!( let vec = b!(
self.hash_variant_identifiers, self.hash_variant_identifiers,
Ok(hash_variant_identifiers Ok(hash_variant_identifiers
.scan_prefix(&hash) .scan_prefix(hash.clone())
.filter_map(|res| res.ok()) .filter_map(|res| res.ok())
.filter_map(|(key, ivec)| { .filter_map(|(key, ivec)| {
let identifier = I::from_bytes(ivec.to_vec()).ok(); let identifier = I::from_bytes(ivec.to_vec()).ok();
@ -1164,8 +1176,10 @@ impl HashRepo for SledRepo {
Ok(vec) Ok(vec)
} }
#[tracing::instrument(level = "trace", skip(self, hash), fields(hash = hex::encode(&hash)))] #[tracing::instrument(level = "trace", skip(self))]
async fn remove_variant(&self, hash: Self::Bytes, variant: String) -> Result<(), RepoError> { async fn remove_variant(&self, hash: Hash, variant: String) -> Result<(), RepoError> {
let hash = hash.to_bytes();
let key = variant_key(&hash, &variant); let key = variant_key(&hash, &variant);
b!( b!(
@ -1176,12 +1190,13 @@ impl HashRepo for SledRepo {
Ok(()) Ok(())
} }
#[tracing::instrument(level = "trace", skip(self, hash, identifier), fields(hash = hex::encode(&hash), identifier = identifier.string_repr()))] #[tracing::instrument(level = "trace", skip(self, identifier), fields(identifier = identifier.string_repr()))]
async fn relate_motion_identifier<I: Identifier>( async fn relate_motion_identifier<I: Identifier>(
&self, &self,
hash: Self::Bytes, hash: Hash,
identifier: &I, identifier: &I,
) -> Result<(), StoreError> { ) -> Result<(), StoreError> {
let hash = hash.to_ivec();
let bytes = identifier.to_bytes()?; let bytes = identifier.to_bytes()?;
b!( b!(
@ -1192,11 +1207,13 @@ impl HashRepo for SledRepo {
Ok(()) Ok(())
} }
#[tracing::instrument(level = "trace", skip(self, hash), fields(hash = hex::encode(&hash)))] #[tracing::instrument(level = "trace", skip(self))]
async fn motion_identifier<I: Identifier + 'static>( async fn motion_identifier<I: Identifier + 'static>(
&self, &self,
hash: Self::Bytes, hash: Hash,
) -> Result<Option<I>, StoreError> { ) -> Result<Option<I>, StoreError> {
let hash = hash.to_ivec();
let opt = b!( let opt = b!(
self.hash_motion_identifiers, self.hash_motion_identifiers,
hash_motion_identifiers.get(hash) hash_motion_identifiers.get(hash)
@ -1205,8 +1222,10 @@ impl HashRepo for SledRepo {
opt.map(|ivec| I::from_bytes(ivec.to_vec())).transpose() opt.map(|ivec| I::from_bytes(ivec.to_vec())).transpose()
} }
#[tracing::instrument(skip(self, hash), fields(hash = hex::encode(&hash)))] #[tracing::instrument(skip(self))]
async fn cleanup(&self, hash: Self::Bytes) -> Result<(), RepoError> { async fn cleanup(&self, hash: Hash) -> Result<(), RepoError> {
let hash = hash.to_ivec();
let hashes = self.hashes.clone(); let hashes = self.hashes.clone();
let hash_identifiers = self.hash_identifiers.clone(); let hash_identifiers = self.hash_identifiers.clone();
let hash_motion_identifiers = self.hash_motion_identifiers.clone(); let hash_motion_identifiers = self.hash_motion_identifiers.clone();
@ -1274,8 +1293,9 @@ impl AliasRepo for SledRepo {
&self, &self,
alias: &Alias, alias: &Alias,
delete_token: &DeleteToken, delete_token: &DeleteToken,
hash: Self::Bytes, hash: Hash,
) -> Result<Result<(), AliasAlreadyExists>, RepoError> { ) -> Result<Result<(), AliasAlreadyExists>, RepoError> {
let hash = hash.to_ivec();
let alias: sled::IVec = alias.to_bytes().into(); let alias: sled::IVec = alias.to_bytes().into();
let delete_token: sled::IVec = delete_token.to_bytes().into(); let delete_token: sled::IVec = delete_token.to_bytes().into();
@ -1328,16 +1348,18 @@ impl AliasRepo for SledRepo {
} }
#[tracing::instrument(level = "trace", skip(self))] #[tracing::instrument(level = "trace", skip(self))]
async fn hash(&self, alias: &Alias) -> Result<Option<Self::Bytes>, RepoError> { async fn hash(&self, alias: &Alias) -> Result<Option<Hash>, RepoError> {
let key = alias.to_bytes(); let key = alias.to_bytes();
let opt = b!(self.alias_hashes, alias_hashes.get(key)); let opt = b!(self.alias_hashes, alias_hashes.get(key));
Ok(opt) Ok(opt.and_then(Hash::from_ivec))
} }
#[tracing::instrument(skip_all)] #[tracing::instrument(skip_all)]
async fn for_hash(&self, hash: Self::Bytes) -> Result<Vec<Alias>, RepoError> { async fn for_hash(&self, hash: Hash) -> Result<Vec<Alias>, RepoError> {
let hash = hash.to_ivec();
let v = b!(self.hash_aliases, { let v = b!(self.hash_aliases, {
Ok(hash_aliases Ok(hash_aliases
.scan_prefix(hash) .scan_prefix(hash)