feltdb 0.4.1

//! One logical durable authority across embedded and local multi-process use.
//!
//! The authority server is the sole process which opens the storage log. IPC
//! clients never mutate storage artifacts and every authoritative read crosses
//! the socket boundary.

use crate::{
    AtomicCommit, AtomicMutation, AtomicPrecondition, FeltDb, PayloadBinding, RecordPrecondition,
    StoredRow,
};
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sha2::{Digest, Sha256};
use std::{
    io,
    path::{Path, PathBuf},
};
use tokio::{
    io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
    net::{UnixListener, UnixStream},
};

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum ReadKind {
    Snapshot,
    Authoritative,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AuthorityRead {
    pub observed_revision: u64,
    pub authoritative_revision: u64,
    pub kind: ReadKind,
    pub content_hash: String,
    pub rows: Vec<StoredRow>,
}

impl AuthorityRead {
    fn current(db: &FeltDb, kind: ReadKind) -> crate::Result<Self> {
        let (revision, mut rows) = db.authority_state();
        rows.sort_by(|left, right| {
            (&left.capability, &left.key).cmp(&(&right.capability, &right.key))
        });
        let mut hasher = Sha256::new();
        hasher.update(revision.to_le_bytes());
        hasher.update(serde_json::to_vec(&rows)?);
        Ok(Self {
            observed_revision: revision,
            authoritative_revision: revision,
            kind,
            content_hash: format!("sha256:{:x}", hasher.finalize()),
            rows,
        })
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AuthorityTransaction {
    pub transaction_id: String,
    /// Explicit optimistic fence. `None` asks the authority to begin and
    /// commit an unconditional transaction at its current revision.
    #[serde(default)]
    pub base_revision: Option<u64>,
    #[serde(default)]
    pub preconditions: Vec<AtomicPrecondition>,
    #[serde(default)]
    pub record_preconditions: Vec<RecordPrecondition>,
    pub mutations: Vec<AtomicMutation>,
    #[serde(default)]
    pub audit: Option<Value>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AuthorityError {
    pub code: String,
    pub message: String,
    #[serde(default)]
    pub authoritative_revision: Option<u64>,
}

impl std::fmt::Display for AuthorityError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(formatter, "{}: {}", self.code, self.message)
    }
}

impl std::error::Error for AuthorityError {}

#[async_trait]
pub trait FeltDbAuthority: Send + Sync {
    async fn read_authoritative(&self) -> Result<AuthorityRead, AuthorityError>;
    async fn current_revision(&self) -> Result<u64, AuthorityError>;
    async fn commit(
        &self,
        transaction: AuthorityTransaction,
    ) -> Result<AtomicCommit, AuthorityError>;
}

#[derive(Clone)]
pub struct EmbeddedAuthority {
    db: FeltDb,
}

impl EmbeddedAuthority {
    pub fn open(path: impl AsRef<Path>) -> crate::Result<Self> {
        Ok(Self {
            db: FeltDb::open(path)?,
        })
    }

    pub fn from_db(db: FeltDb) -> Self {
        Self { db }
    }

    /// An explicitly historical value. It never claims freshness beyond its
    /// captured `observed_revision`.
    pub fn snapshot(&self) -> crate::Result<AuthorityRead> {
        AuthorityRead::current(&self.db, ReadKind::Snapshot)
    }
}

/// Names the construction of the payload digest the authority binds ids to.
///
/// The prefix is stored with the digest in the durable transaction record, so a
/// digest written by a different construction is recognizable as one this
/// build cannot judge, instead of being read as a mismatch.
const PAYLOAD_DIGEST_PREFIX: &str = "authority-tx1:sha256:";

/// Canonical text of a JSON value: object members sorted by name at every
/// depth, array order preserved. Written out here rather than left to
/// `serde_json`'s map ordering, which is a build feature that another crate in
/// the dependency graph can change.
fn canonical_json(value: &Value, out: &mut String) {
    match value {
        Value::Object(map) => {
            let mut keys: Vec<&String> = map.keys().collect();
            keys.sort();
            out.push('{');
            for (index, key) in keys.into_iter().enumerate() {
                if index > 0 {
                    out.push(',');
                }
                out.push_str(&Value::String(key.clone()).to_string());
                out.push(':');
                canonical_json(&map[key], out);
            }
            out.push('}');
        }
        Value::Array(items) => {
            out.push('[');
            for (index, item) in items.iter().enumerate() {
                if index > 0 {
                    out.push(',');
                }
                canonical_json(item, out);
            }
            out.push(']');
        }
        other => out.push_str(&other.to_string()),
    }
}

/// The digest an id is bound to: everything the commit acts on or records, and
/// not the id itself. `audit` is included because it is committed evidence the
/// caller supplied, not transport metadata; two requests that record different
/// audit facts under one id are different requests.
fn payload_digest(transaction: &AuthorityTransaction) -> Result<String, AuthorityError> {
    let semantic = serde_json::json!({
        "base_revision": transaction.base_revision,
        "preconditions": transaction.preconditions,
        "record_preconditions": transaction.record_preconditions,
        "mutations": transaction.mutations,
        "audit": transaction.audit,
    });
    let mut canonical = String::new();
    canonical_json(&semantic, &mut canonical);
    Ok(format!(
        "{PAYLOAD_DIGEST_PREFIX}{:x}",
        Sha256::digest(canonical.as_bytes())
    ))
}

fn authority_error(error: crate::FlowError) -> AuthorityError {
    match error {
        crate::FlowError::TransactionIdPayloadMismatch { transaction_id: id } => {
            AuthorityError {
                code: "IDEMPOTENCY_CONFLICT".into(),
                message: format!(
                    "transaction {id} was already committed with a different payload"
                ),
                authoritative_revision: None,
            }
        }
        crate::FlowError::RevisionConflict {
            expected: _,
            actual,
        } => AuthorityError {
            code: "REVISION_CONFLICT".into(),
            message: error.to_string(),
            authoritative_revision: Some(actual),
        },
        other => AuthorityError {
            code: "AUTHORITY_ERROR".into(),
            message: other.to_string(),
            authoritative_revision: None,
        },
    }
}

#[async_trait]
impl FeltDbAuthority for EmbeddedAuthority {
    async fn read_authoritative(&self) -> Result<AuthorityRead, AuthorityError> {
        AuthorityRead::current(&self.db, ReadKind::Authoritative).map_err(authority_error)
    }

    async fn current_revision(&self) -> Result<u64, AuthorityError> {
        self.db.current_revision().map_err(authority_error)
    }

    async fn commit(
        &self,
        transaction: AuthorityTransaction,
    ) -> Result<AtomicCommit, AuthorityError> {
        let digest = payload_digest(&transaction)?;
        // How the id is already known, if it is. A recorded hash never changes
        // and a missing one never appears, so this read cannot be invalidated
        // by a concurrent commit: the comparison that matters -- the id against
        // the digest of this request -- is made by the core, atomically with
        // the commit, under the lock that serializes every writer.
        let id = transaction.transaction_id.as_str();
        let applied = self.db.has_applied_transaction(id).map_err(authority_error)?;
        let recorded = if applied {
            self.db
                .applied_transaction_payload_hash(id)
                .map_err(authority_error)?
        } else {
            None
        };
        let (bound, legacy) = match (applied, recorded) {
            (false, _) => (Some(digest), false),
            // Committed before ids were bound: the id is all there is to compare.
            (true, None) => (None, true),
            (true, Some(hash)) if hash.starts_with(PAYLOAD_DIGEST_PREFIX) => (Some(digest), false),
            // Recorded by a construction this build does not know, or by another
            // writer's: neither a match nor a mismatch can be established.
            (true, Some(_)) => {
                return Err(AuthorityError {
                    code: "IDEMPOTENCY_UNVERIFIABLE".into(),
                    message: format!(
                        "transaction {id} exists, but its recorded payload identity cannot be verified by this build"
                    ),
                    authoritative_revision: None,
                })
            }
        };
        let mut commit = self
            .db
            .apply_atomic_transaction_guarded(
                id,
                bound.as_deref(),
                transaction.base_revision,
                &transaction.preconditions,
                &transaction.record_preconditions,
                &transaction.mutations,
                transaction.audit.clone(),
            )
            .map_err(authority_error)?;
        if legacy && commit.duplicate {
            commit.payload_binding = Some(PayloadBinding::LegacyUnverified);
        }
        Ok(commit)
    }
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "operation", rename_all = "snake_case")]
enum Request {
    Read,
    CurrentRevision,
    Commit { transaction: AuthorityTransaction },
}

#[derive(Debug, Serialize, Deserialize)]
#[serde(tag = "result", content = "value", rename_all = "snake_case")]
enum Response {
    Read(AuthorityRead),
    Revision(u64),
    Commit(AtomicCommit),
    Error(AuthorityError),
}

pub struct AuthorityServer {
    socket_path: PathBuf,
    authority: EmbeddedAuthority,
}

impl AuthorityServer {
    pub fn open(
        socket_path: impl Into<PathBuf>,
        storage_path: impl AsRef<Path>,
    ) -> crate::Result<Self> {
        Ok(Self {
            socket_path: socket_path.into(),
            authority: EmbeddedAuthority::open(storage_path)?,
        })
    }

    pub async fn serve(self) -> io::Result<()> {
        if let Some(parent) = self.socket_path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        if self.socket_path.exists() {
            use std::os::unix::fs::FileTypeExt;
            if !std::fs::symlink_metadata(&self.socket_path)?
                .file_type()
                .is_socket()
            {
                return Err(io::Error::new(
                    io::ErrorKind::AlreadyExists,
                    "authority socket path is not a socket",
                ));
            }
            std::fs::remove_file(&self.socket_path)?;
        }
        let listener = UnixListener::bind(&self.socket_path)?;
        loop {
            let (stream, _) = listener.accept().await?;
            let authority = self.authority.clone();
            tokio::spawn(async move {
                let _ = handle_connection(stream, authority).await;
            });
        }
    }
}

async fn handle_connection(stream: UnixStream, authority: EmbeddedAuthority) -> io::Result<()> {
    let (reader, mut writer) = stream.into_split();
    let mut lines = BufReader::new(reader).lines();
    while let Some(line) = lines.next_line().await? {
        let response = match serde_json::from_str::<Request>(&line) {
            Ok(Request::Read) => authority.read_authoritative().await.map(Response::Read),
            Ok(Request::CurrentRevision) => {
                authority.current_revision().await.map(Response::Revision)
            }
            Ok(Request::Commit { transaction }) => {
                authority.commit(transaction).await.map(Response::Commit)
            }
            Err(error) => Err(AuthorityError {
                code: "INVALID_REQUEST".into(),
                message: error.to_string(),
                authoritative_revision: None,
            }),
        }
        .unwrap_or_else(Response::Error);
        writer
            .write_all(&serde_json::to_vec(&response).map_err(io::Error::other)?)
            .await?;
        writer.write_all(b"\n").await?;
    }
    Ok(())
}

#[derive(Clone)]
pub struct AuthorityClient {
    socket_path: PathBuf,
}

impl AuthorityClient {
    pub fn connect(socket_path: impl Into<PathBuf>) -> Self {
        Self {
            socket_path: socket_path.into(),
        }
    }

    async fn request(&self, request: Request) -> Result<Response, AuthorityError> {
        // Once a commit has been handed to the authority it may have been
        // applied whether or not an answer comes back, so every failure from
        // that point is an unknown outcome, not a failure.
        let is_commit = matches!(request, Request::Commit { .. });
        let unknown = |detail: String| AuthorityError {
            code: "COMMIT_OUTCOME_UNKNOWN".into(),
            message: format!(
                "the commit may or may not have been applied ({detail}); retry with the same transaction id and payload"
            ),
            authoritative_revision: None,
        };
        let mut stream = UnixStream::connect(&self.socket_path)
            .await
            .map_err(|error| AuthorityError {
                code: "AUTHORITY_UNAVAILABLE".into(),
                message: error.to_string(),
                authoritative_revision: None,
            })?;
        stream
            .write_all(
                &serde_json::to_vec(&request).map_err(|error| AuthorityError {
                    code: "INVALID_REQUEST".into(),
                    message: error.to_string(),
                    authoritative_revision: None,
                })?,
            )
            .await
            .map_err(|error| if is_commit { unknown(error.to_string()) } else { io_authority_error(error) })?;
        stream
            .write_all(b"\n")
            .await
            .map_err(|error| if is_commit { unknown(error.to_string()) } else { io_authority_error(error) })?;
        let mut response = String::new();
        BufReader::new(stream)
            .read_line(&mut response)
            .await
            .map_err(|error| if is_commit { unknown(error.to_string()) } else { io_authority_error(error) })?;
        match serde_json::from_str(&response).map_err(|error| {
            if is_commit {
                unknown(format!("unreadable response: {error}"))
            } else {
                AuthorityError {
                    code: "INVALID_RESPONSE".into(),
                    message: error.to_string(),
                    authoritative_revision: None,
                }
            }
        })? {
            Response::Error(error) => Err(error),
            response => Ok(response),
        }
    }
}

fn io_authority_error(error: io::Error) -> AuthorityError {
    AuthorityError {
        code: "AUTHORITY_UNAVAILABLE".into(),
        message: error.to_string(),
        authoritative_revision: None,
    }
}

#[async_trait]
impl FeltDbAuthority for AuthorityClient {
    async fn read_authoritative(&self) -> Result<AuthorityRead, AuthorityError> {
        match self.request(Request::Read).await? {
            Response::Read(read) => Ok(read),
            _ => Err(AuthorityError {
                code: "INVALID_RESPONSE".into(),
                message: "expected authority read".into(),
                authoritative_revision: None,
            }),
        }
    }

    async fn current_revision(&self) -> Result<u64, AuthorityError> {
        match self.request(Request::CurrentRevision).await? {
            Response::Revision(revision) => Ok(revision),
            _ => Err(AuthorityError {
                code: "INVALID_RESPONSE".into(),
                message: "expected authority revision".into(),
                authoritative_revision: None,
            }),
        }
    }

    async fn commit(
        &self,
        transaction: AuthorityTransaction,
    ) -> Result<AtomicCommit, AuthorityError> {
        match self.request(Request::Commit { transaction }).await? {
            Response::Commit(commit) => Ok(commit),
            _ => Err(AuthorityError {
                code: "COMMIT_OUTCOME_UNKNOWN".into(),
                message: "the authority answered a commit with something other than a commit; \
                          its outcome is unknown, retry with the same transaction id and payload"
                    .into(),
                authoritative_revision: None,
            }),
        }
    }
}