did-hosting-common 0.10.1

//! Server-local domain assignment cache (T28 + T29 staging).
//!
//! Each running server keeps a local record of which domains it is
//! currently authoritative for. The list is populated by inbound
//! `replica/domain/assign` / `replica/domain/unassign` Trust Tasks from the
//! control plane (T28) and read on cold start before the control
//! plane is reachable (T29's fallback chain).
//!
//! ## Storage shape
//!
//! - **Key**: `assignments:<domain>` — one entry per assigned domain.
//!   The keyspace const `KS_ASSIGNMENTS` carries the prefix; per-
//!   record keys are built via [`assignment_key`].
//! - **Value**: [`AssignmentEntry`] — JSON-serialised.
//!
//! ## Idempotency semantics
//!
//! - **assign(d)** when `d` already present → returns `Existing`. No
//!   timestamp / assigner mutation; the original `assigned_at` is
//!   preserved so audit-log traces stay stable.
//! - **assign(d)** when `d` absent → returns `Created` and writes the
//!   entry.
//! - **unassign(d)** when `d` present → returns `Removed`.
//! - **unassign(d)** when `d` absent → returns `Missing`. No-op for
//!   the storage layer (the caller decides whether to audit-log the
//!   noise).

use serde::{Deserialize, Serialize};

use super::error::AppError;
use super::pending_purge::{self, PendingPurge};
use super::store::{KS_ASSIGNMENTS, Store};

/// One row in the per-server assignment cache.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct AssignmentEntry {
    /// Normalised domain name (lowercase, IDNA-encoded). Matches the
    /// [`super::domain::DomainEntry::name`] form so cross-key joins
    /// work without re-normalisation.
    pub domain: String,
    /// Epoch second of the first successful assignment. Preserved
    /// across re-assigns for audit-log stability.
    pub assigned_at: u64,
    /// The control-plane DID that issued the assignment. Stored so an
    /// operator can answer "which control plane told me to host this?"
    /// during multi-control-plane migrations.
    pub assigner: String,
}

/// Build the storage key for `domain`. The keyspace const is
/// `KS_ASSIGNMENTS = "assignments"`; per-record keys are
/// `assignments:<domain>`.
fn assignment_key(domain: &str) -> String {
    format!("assignments:{domain}")
}

/// Outcome of an idempotent [`assign`] call.
#[derive(Debug, PartialEq, Eq)]
pub enum AssignOutcome {
    /// Entry was created — caller should audit-log the event.
    Created(AssignmentEntry),
    /// Entry already existed — caller should suppress duplicate audit
    /// noise. The existing entry is returned for inspection.
    Existing(AssignmentEntry),
}

/// Outcome of an idempotent [`unassign`] call.
#[derive(Debug, PartialEq, Eq)]
pub enum UnassignOutcome {
    /// Entry was present and has been removed.
    Removed(AssignmentEntry),
    /// Entry was already absent — no-op. Caller suppresses audit noise.
    Missing,
}

/// Assign `domain` to this server. Idempotent — re-assigning a known
/// domain returns [`AssignOutcome::Existing`] without mutating the
/// stored row.
pub async fn assign(
    store: &Store,
    domain: &str,
    assigner: &str,
    now_epoch: u64,
) -> Result<AssignOutcome, AppError> {
    let ks = store.keyspace(KS_ASSIGNMENTS)?;
    let key = assignment_key(domain);

    if let Some(existing) = ks.get::<AssignmentEntry>(key.clone()).await? {
        return Ok(AssignOutcome::Existing(existing));
    }

    let entry = AssignmentEntry {
        domain: domain.to_string(),
        assigned_at: now_epoch,
        assigner: assigner.to_string(),
    };
    ks.insert(key, &entry).await?;
    Ok(AssignOutcome::Created(entry))
}

/// Record a control plane's assignment of `domain` to this server, at
/// `now_epoch` — also when the domain is already assigned.
///
/// This is `replica/domain/assign/0.1`'s rule: a re-assignment *refreshes* the
/// assignment time, because that time is what `replica/domain/purge/0.1`'s
/// freshness check compares a purge's `issuedAt` against. A purge issued before
/// the operator re-assigned the domain must never wipe it. (The first-boot seed
/// uses [`assign`], which keeps an existing row as it is.)
pub async fn record_assignment(
    store: &Store,
    domain: &str,
    assigner: &str,
    now_epoch: u64,
) -> Result<AssignmentEntry, AppError> {
    let entry = AssignmentEntry {
        domain: domain.to_string(),
        assigned_at: now_epoch,
        assigner: assigner.to_string(),
    };
    store
        .keyspace(KS_ASSIGNMENTS)?
        .insert(assignment_key(domain), &entry)
        .await?;
    Ok(entry)
}

/// [`record_assignment`] and cancel any purge scheduled for `domain`, in one
/// atomic write — `replica/domain/assign`'s whole effect on the store.
///
/// Written as two, a crash after the assignment but before the cancel left
/// the domain assigned with a ripe purge still queued, which the sweep would
/// carry out against the domain the operator had just taken back. Returns
/// the entry written and the purge that was cancelled, if there was one.
pub async fn record_assignment_cancelling_purge(
    store: &Store,
    domain: &str,
    assigner: &str,
    now_epoch: u64,
) -> Result<(AssignmentEntry, Option<PendingPurge>), AppError> {
    let ks = store.keyspace(KS_ASSIGNMENTS)?;
    let cancelled = pending_purge::get(store, domain).await?;
    let entry = AssignmentEntry {
        domain: domain.to_string(),
        assigned_at: now_epoch,
        assigner: assigner.to_string(),
    };
    let mut batch = store.batch();
    batch.insert(&ks, assignment_key(domain), &entry)?;
    pending_purge::stage_cancel(store, &mut batch, domain)?;
    batch.commit().await?;
    Ok((entry, cancelled))
}

/// [`unassign`] and, when an assignment was removed, schedule the domain's
/// purge — in one atomic write. Written as two, a crash between them left
/// the domain unassigned with no purge ever coming, after the directive had
/// been answered `scheduled`. Returns the outcome and the schedule written.
pub async fn unassign_scheduling_purge(
    store: &Store,
    domain: &str,
    now_epoch: u64,
    grace_seconds: u64,
    reason: &str,
    scheduled_by: &str,
) -> Result<(UnassignOutcome, Option<PendingPurge>), AppError> {
    let ks = store.keyspace(KS_ASSIGNMENTS)?;
    let key = assignment_key(domain);
    let Some(existing) = ks.get::<AssignmentEntry>(key.clone()).await? else {
        return Ok((UnassignOutcome::Missing, None));
    };
    let mut batch = store.batch();
    batch.remove(&ks, key);
    let scheduled = pending_purge::stage_schedule(
        store,
        &mut batch,
        domain,
        now_epoch,
        grace_seconds,
        reason,
        scheduled_by,
    )?;
    batch.commit().await?;
    Ok((UnassignOutcome::Removed(existing), Some(scheduled)))
}

/// Unassign `domain` from this server. Idempotent.
pub async fn unassign(store: &Store, domain: &str) -> Result<UnassignOutcome, AppError> {
    let ks = store.keyspace(KS_ASSIGNMENTS)?;
    let key = assignment_key(domain);

    let Some(existing) = ks.get::<AssignmentEntry>(key.clone()).await? else {
        return Ok(UnassignOutcome::Missing);
    };
    ks.remove(key).await?;
    Ok(UnassignOutcome::Removed(existing))
}

/// Look up a single assignment by domain.
pub async fn get(store: &Store, domain: &str) -> Result<Option<AssignmentEntry>, AppError> {
    let ks = store.keyspace(KS_ASSIGNMENTS)?;
    ks.get(assignment_key(domain)).await
}

/// List every assignment held by this server. Read on cold-start by
/// T29's fallback chain.
pub async fn list(store: &Store) -> Result<Vec<AssignmentEntry>, AppError> {
    let ks = store.keyspace(KS_ASSIGNMENTS)?;
    let raw = ks.prefix_iter_raw(b"assignments:".to_vec()).await?;
    let mut out = Vec::with_capacity(raw.len());
    for (_k, v) in raw {
        if let Ok(entry) = serde_json::from_slice::<AssignmentEntry>(&v) {
            out.push(entry);
        }
    }
    Ok(out)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::server::config::StoreConfig;

    async fn fjall_store() -> Store {
        let dir = tempfile::tempdir().expect("tempdir");
        let cfg = StoreConfig {
            data_dir: dir.path().to_path_buf(),
            ..StoreConfig::default()
        };
        std::mem::forget(dir);
        Store::open(&cfg).await.expect("open fjall")
    }

    #[tokio::test]
    async fn assign_creates_then_idempotent() {
        let store = fjall_store().await;
        let first = assign(&store, "example.com", "did:example:control", 100)
            .await
            .unwrap();
        assert!(matches!(first, AssignOutcome::Created(_)));

        // Same call again — Existing, same assigned_at (no clobber).
        let second = assign(&store, "example.com", "did:example:control", 200)
            .await
            .unwrap();
        let AssignOutcome::Existing(entry) = second else {
            panic!("expected Existing on re-assign, got {second:?}");
        };
        assert_eq!(
            entry.assigned_at, 100,
            "re-assign must preserve original timestamp"
        );
        assert_eq!(entry.assigner, "did:example:control");
    }

    #[tokio::test]
    async fn a_recorded_assignment_refreshes_its_time() {
        let store = fjall_store().await;
        record_assignment(&store, "example.com", "did:example:control", 100)
            .await
            .unwrap();
        record_assignment(&store, "example.com", "did:example:control", 200)
            .await
            .unwrap();
        let entry = get(&store, "example.com").await.unwrap().unwrap();
        assert_eq!(entry.assigned_at, 200);
    }

    #[tokio::test]
    async fn unassign_round_trip() {
        let store = fjall_store().await;
        // Unassign missing is a no-op.
        let missing = unassign(&store, "example.com").await.unwrap();
        assert!(matches!(missing, UnassignOutcome::Missing));

        // Assign then unassign.
        assign(&store, "example.com", "did:example:control", 1)
            .await
            .unwrap();
        let removed = unassign(&store, "example.com").await.unwrap();
        let UnassignOutcome::Removed(entry) = removed else {
            panic!("expected Removed, got {removed:?}");
        };
        assert_eq!(entry.domain, "example.com");

        // Second unassign is Missing again.
        let missing2 = unassign(&store, "example.com").await.unwrap();
        assert!(matches!(missing2, UnassignOutcome::Missing));
    }

    #[tokio::test]
    async fn list_returns_every_assigned_domain() {
        let store = fjall_store().await;
        assign(&store, "a.example", "did:c", 1).await.unwrap();
        assign(&store, "b.example", "did:c", 1).await.unwrap();
        assign(&store, "c.example", "did:c", 1).await.unwrap();
        let all = list(&store).await.unwrap();
        let mut names: Vec<String> = all.into_iter().map(|e| e.domain).collect();
        names.sort();
        assert_eq!(names, vec!["a.example", "b.example", "c.example"]);
    }

    #[tokio::test]
    async fn get_returns_none_for_unknown() {
        let store = fjall_store().await;
        assert!(get(&store, "unknown.example").await.unwrap().is_none());
    }

    /// Crash-ordering guard for `replica/domain/assign`: an assigned domain
    /// must never be observable with its purge still queued — the state a
    /// crash between two separate writes would leave for the sweep to act
    /// on. Read assignment-then-purge so an atomic write cannot false-alarm.
    #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
    async fn assignment_never_leaves_a_purge_queued() {
        use crate::server::store::test_support::assert_never_observed;
        const N: usize = 150;
        let store = fjall_store().await;
        for i in 0..N {
            pending_purge::schedule(
                &store,
                &format!("d{i}.example"),
                1,
                60,
                "grace-expired",
                "did:c",
            )
            .await
            .unwrap();
        }
        let check_store = store.clone();
        assert_never_observed(
            N,
            |i| {
                let store = store.clone();
                async move {
                    let (_, cancelled) = record_assignment_cancelling_purge(
                        &store,
                        &format!("d{i}.example"),
                        "did:c",
                        100,
                    )
                    .await
                    .unwrap();
                    assert!(cancelled.is_some());
                }
            },
            move |i| {
                let store = check_store.clone();
                async move {
                    let domain = format!("d{i}.example");
                    get(&store, &domain).await.unwrap()?;
                    pending_purge::get(&store, &domain)
                        .await
                        .unwrap()
                        .map(|_| format!("{domain} assigned with a purge queued"))
                }
            },
        )
        .await;
    }

    /// Crash-ordering guard for `replica/domain/unassign`: once the
    /// assignment is gone the purge must already be scheduled.
    #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
    async fn unassignment_never_leaves_a_domain_unscheduled() {
        use crate::server::store::test_support::assert_never_observed;
        const N: usize = 150;
        let store = fjall_store().await;
        for i in 0..N {
            record_assignment(&store, &format!("d{i}.example"), "did:c", 1)
                .await
                .unwrap();
        }
        let check_store = store.clone();
        assert_never_observed(
            N,
            |i| {
                let store = store.clone();
                async move {
                    let (outcome, scheduled) = unassign_scheduling_purge(
                        &store,
                        &format!("d{i}.example"),
                        100,
                        60,
                        "grace-expired",
                        "did:c",
                    )
                    .await
                    .unwrap();
                    assert!(matches!(outcome, UnassignOutcome::Removed(_)));
                    assert!(scheduled.is_some());
                }
            },
            move |i| {
                let store = check_store.clone();
                async move {
                    let domain = format!("d{i}.example");
                    if get(&store, &domain).await.unwrap().is_some() {
                        return None;
                    }
                    match pending_purge::get(&store, &domain).await.unwrap() {
                        Some(_) => None,
                        None => Some(format!("{domain} unassigned with no purge")),
                    }
                }
            },
        )
        .await;
    }

    #[tokio::test]
    async fn unassigning_an_unassigned_domain_schedules_nothing() {
        let store = fjall_store().await;
        let (outcome, scheduled) =
            unassign_scheduling_purge(&store, "x.example", 1, 60, "grace-expired", "did:c")
                .await
                .unwrap();
        assert_eq!(outcome, UnassignOutcome::Missing);
        assert!(scheduled.is_none());
        assert!(
            pending_purge::get(&store, "x.example")
                .await
                .unwrap()
                .is_none()
        );
    }
}