Monitors and federation server side (#225)

Implements the server-side for the following issues:
- #224
- #227
- #228
- #232 

Deferred #229 for a follow up PR, due to the large diff.

crates/api/src/v1/fetch.rs

@@ -6,7 +6,7 @@ use std::{borrow::Cow, collections::HashMap};
 use thiserror::Error;
 use warg_crypto::hash::AnyHash;
 use warg_protocol::{
-    registry::{LogId, RegistryLen},
+    registry::{LogId, PackageId, RegistryLen},
     PublishedProtoEnvelopeBody,
 };
 
@@ -53,6 +53,24 @@ pub struct FetchLogsResponse {
     pub packages: HashMap<LogId, Vec<PublishedRecord>>,
 }
 
+/// Represents a fetch package names request.
+#[derive(Debug, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct FetchPackageNamesRequest<'a> {
+    /// List of package log IDs to request the package name.
+    pub packages: Cow<'a, Vec<LogId>>,
+}
+
+/// Represents a fetch package names response. If the requested number of packages exceeds the limit
+/// that the server can fulfill on a single request, the client should retry with the log IDs that
+/// are absent in the response body.
+#[derive(Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct FetchPackageNamesResponse {
+    /// The log ID hash mapping to a package ID. If `None`, the package name cannot be provided.
+    pub packages: HashMap<LogId, Option<PackageId>>,
+}
+
 /// Represents a fetch API error.
 #[non_exhaustive]
 #[derive(Debug, Error)]

crates/api/src/v1/ledger.rs

@@ -0,0 +1,113 @@
+//! Types relating to the ledger API.
+
+use serde::{Deserialize, Serialize, Serializer};
+use std::borrow::Cow;
+use thiserror::Error;
+use warg_crypto::hash::HashAlgorithm;
+use warg_protocol::registry::RegistryIndex;
+
+/// Represents response a get ledger sources request.
+#[derive(Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct LedgerSourcesResponse {
+    /// The hash algorithm used by the ledger.
+    pub hash_algorithm: HashAlgorithm,
+    /// The list of ledger sources.
+    pub sources: Vec<LedgerSource>,
+}
+
+/// Ledger source for a specified registry index range. Expected to be sorted in ascending order.
+#[derive(Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct LedgerSource {
+    /// First registry index that is included in this ledger source.
+    pub first_registry_index: RegistryIndex,
+    /// Last registry index that is included in this ledger source.
+    pub last_registry_index: RegistryIndex,
+    /// The HTTP GET URL location for the ledger source.
+    pub url: String,
+    /// Content type for the ledger source.
+    pub content_type: LedgerSourceContentType,
+    /// Optional, server accepts for HTTP Range header.
+    #[serde(default, skip_serializing_if = "is_false")]
+    pub accept_ranges: bool,
+}
+
+fn is_false(b: &bool) -> bool {
+    !b
+}
+
+/// Content type for the ledger source.
+#[derive(Default, PartialEq, Serialize, Deserialize, Debug)]
+pub enum LedgerSourceContentType {
+    /// The content type is binary representation of the LogId and RecordId hashes without padding.
+    /// In the case of `sha256` hash algorithm, this is a repeating sequence of 64 bytes (32 bytes
+    /// for each the LogId and RecordId) without padding.
+    #[default]
+    #[serde(rename = "application/vnd.warg.ledger.packed")]
+    Packed,
+}
+
+impl LedgerSourceContentType {
+    /// Returns the content type represented as a string.
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            Self::Packed => "application/vnd.warg.ledger.packed",
+        }
+    }
+}
+
+/// Represents a ledger API error.
+#[non_exhaustive]
+#[derive(Debug, Error)]
+pub enum LedgerError {
+    /// An error with a message occurred.
+    #[error("{message}")]
+    Message {
+        /// The HTTP status code.
+        status: u16,
+        /// The error message
+        message: String,
+    },
+}
+
+impl LedgerError {
+    /// Returns the HTTP status code of the error.
+    pub fn status(&self) -> u16 {
+        match self {
+            Self::Message { status, .. } => *status,
+        }
+    }
+}
+
+#[derive(Serialize, Deserialize)]
+#[serde(untagged, rename_all = "camelCase")]
+enum RawError<'a> {
+    Message { status: u16, message: Cow<'a, str> },
+}
+
+impl Serialize for LedgerError {
+    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        match self {
+            Self::Message { status, message } => RawError::Message {
+                status: *status,
+                message: Cow::Borrowed(message),
+            }
+            .serialize(serializer),
+        }
+    }
+}
+
+impl<'de> Deserialize<'de> for LedgerError {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        match RawError::deserialize(deserializer)? {
+            RawError::Message { status, message } => Ok(Self::Message {
+                status,
+                message: message.into_owned(),
+            }),
+        }
+    }
+}

crates/api/src/v1/mod.rs

@@ -2,12 +2,19 @@
 
 pub mod content;
 pub mod fetch;
+pub mod ledger;
+pub mod monitor;
 pub mod package;
 pub mod paths;
 pub mod proof;
 
 use serde::{Deserialize, Serialize};
 
+/// The HTTP request and response header name that specifies the registry domain whose data is the
+/// subject of the request. This header is only expected to be used if referring to a different
+/// registry than the host registry.
+pub const REGISTRY_HEADER_NAME: &str = "warg-registry";
+
 /// Represents the supported kinds of content sources.
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(tag = "type", rename_all = "camelCase")]

crates/api/src/v1/monitor.rs

@@ -0,0 +1,88 @@
+//! Types relating to the monitor API.
+
+use serde::{Deserialize, Serialize, Serializer};
+use std::borrow::Cow;
+use thiserror::Error;
+
+/// Represents checkpoint verification response.
+#[derive(Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CheckpointVerificationResponse {
+    /// The checkpoint verification state.
+    pub checkpoint: VerificationState,
+    /// The checkpoint signature verification state.
+    pub signature: VerificationState,
+    /// Optional, retry after specified number of seconds.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub retry_after: Option<u16>,
+}
+
+/// Represents checkpoint verification state.
+#[derive(Eq, PartialEq, Debug, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum VerificationState {
+    /// The checkpoint is unverified and could be valid or invalid.
+    #[serde(rename_all = "camelCase")]
+    Unverified,
+    /// The checkpoint is verified.
+    #[serde(rename_all = "camelCase")]
+    Verified,
+    /// The checkpoint is invalid.
+    #[serde(rename_all = "camelCase")]
+    Invalid,
+}
+
+/// Represents a monitor API error.
+#[non_exhaustive]
+#[derive(Debug, Error)]
+pub enum MonitorError {
+    /// An error with a message occurred.
+    #[error("{message}")]
+    Message {
+        /// The HTTP status code.
+        status: u16,
+        /// The error message
+        message: String,
+    },
+}
+
+impl MonitorError {
+    /// Returns the HTTP status code of the error.
+    pub fn status(&self) -> u16 {
+        match self {
+            Self::Message { status, .. } => *status,
+        }
+    }
+}
+
+#[derive(Serialize, Deserialize)]
+#[serde(untagged, rename_all = "camelCase")]
+enum RawError<'a> {
+    Message { status: u16, message: Cow<'a, str> },
+}
+
+impl Serialize for MonitorError {
+    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        match self {
+            Self::Message { status, message } => RawError::Message {
+                status: *status,
+                message: Cow::Borrowed(message),
+            }
+            .serialize(serializer),
+        }
+    }
+}
+
+impl<'de> Deserialize<'de> for MonitorError {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        match RawError::deserialize(deserializer)? {
+            RawError::Message { status, message } => Ok(Self::Message {
+                status,
+                message: message.into_owned(),
+            }),
+        }
+    }
+}

crates/api/src/v1/package.rs

@@ -41,7 +41,8 @@ pub struct MissingContent {
 #[derive(Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct PublishRecordRequest<'a> {
-    /// The id of the package being published.
+    /// The package name being published.
+    #[serde(alias = "packageName")]
     pub package_id: Cow<'a, PackageId>,
     /// The publish record to add to the package log.
     pub record: Cow<'a, ProtoEnvelopeBody>,
@@ -120,6 +121,18 @@ pub enum PackageError {
     /// The record is not currently sourcing content.
     #[error("the record is not currently sourcing content")]
     RecordNotSourcing,
+    /// The provided package's namespace was not found in the operator log.
+    #[error("namespace `{0}` is not defined on the registry")]
+    NamespaceNotDefined(String),
+    /// The provided package's namespace is imported from another registry.
+    #[error("namespace `{0}` is an imported namespace from another registry")]
+    NamespaceImported(String),
+    /// The provided package's namespace conflicts with an existing namespace where the name only differs by case.
+    #[error("namespace conflicts with existing namespace `{0}`; package namespaces must be unique in a case insensitive way")]
+    NamespaceConflict(String),
+    /// The provided package name conflicts with an existing package where the name only differs by case.
+    #[error("the package conflicts with existing package name `{0}`; package names must be unique in a case insensitive way")]
+    PackageNameConflict(PackageId),
     /// The operation was not authorized by the registry.
     #[error("unauthorized operation: {0}")]
     Unauthorized(String),
@@ -146,7 +159,10 @@ impl PackageError {
             // Note: this is 403 and not a 401 as the registry does not use
             // HTTP authentication.
             Self::Unauthorized { .. } => 403,
-            Self::LogNotFound(_) | Self::RecordNotFound(_) => 404,
+            Self::LogNotFound(_) | Self::RecordNotFound(_) | Self::NamespaceNotDefined(_) => 404,
+            Self::NamespaceImported(_)
+            | Self::NamespaceConflict(_)
+            | Self::PackageNameConflict(_) => 409,
             Self::RecordNotSourcing => 405,
             Self::Rejection(_) => 422,
             Self::NotSupported(_) => 501,
@@ -160,6 +176,9 @@ impl PackageError {
 enum EntityType {
     Log,
     Record,
+    Namespace,
+    NamespaceImport,
+    Name,
 }
 
 #[derive(Serialize, Deserialize)]
@@ -179,6 +198,12 @@ where
         ty: EntityType,
         id: Cow<'a, T>,
     },
+    Conflict {
+        status: Status<409>,
+        #[serde(rename = "type")]
+        ty: EntityType,
+        id: Cow<'a, T>,
+    },
     RecordNotSourcing {
         status: Status<405>,
     },
@@ -216,6 +241,30 @@ impl Serialize for PackageError {
                 id: Cow::Borrowed(record_id),
             }
             .serialize(serializer),
+            Self::NamespaceNotDefined(namespace) => RawError::NotFound {
+                status: Status::<404>,
+                ty: EntityType::Namespace,
+                id: Cow::Borrowed(namespace),
+            }
+            .serialize(serializer),
+            Self::NamespaceImported(namespace) => RawError::Conflict {
+                status: Status::<409>,
+                ty: EntityType::NamespaceImport,
+                id: Cow::Borrowed(namespace),
+            }
+            .serialize(serializer),
+            Self::NamespaceConflict(existing) => RawError::Conflict {
+                status: Status::<409>,
+                ty: EntityType::Namespace,
+                id: Cow::Borrowed(existing),
+            }
+            .serialize(serializer),
+            Self::PackageNameConflict(existing) => RawError::Conflict {
+                status: Status::<409>,
+                ty: EntityType::Name,
+                id: Cow::Borrowed(existing),
+            }
+            .serialize(serializer),
             Self::RecordNotSourcing => RawError::RecordNotSourcing::<()> {
                 status: Status::<405>,
             }
@@ -266,6 +315,22 @@ impl<'de> Deserialize<'de> for PackageError {
                         })?
                         .into(),
                 )),
+                EntityType::Namespace => Ok(Self::NamespaceNotDefined(id.into_owned())),
+                _ => Err(serde::de::Error::invalid_value(
+                    Unexpected::Enum,
+                    &"a valid entity type",
+                )),
+            },
+            RawError::Conflict { status: _, ty, id } => match ty {
+                EntityType::Namespace => Ok(Self::NamespaceConflict(id.into_owned())),
+                EntityType::NamespaceImport => Ok(Self::NamespaceImported(id.into_owned())),
+                EntityType::Name => Ok(Self::PackageNameConflict(
+                    PackageId::new(id.into_owned()).unwrap(),
+                )),
+                _ => Err(serde::de::Error::invalid_value(
+                    Unexpected::Enum,
+                    &"a valid entity type",
+                )),
             },
             RawError::RecordNotSourcing { status: _ } => Ok(Self::RecordNotSourcing),
             RawError::Rejection { status: _, message } => Ok(Self::Rejection(message.into_owned())),