mirror of
https://relay.ngit.dev/npub15qydau2hjma6ngxkl2cyar74wzyjshvl65za5k5rl69264ar2exs5cyejr/ngit-grasp.git
synced 2026-10-06 15:38:25 +00:00
DanConwayDev
07fe32b704
GRASP-03 must retain the profile and NIP-65 relay list needed by clients that use the GRASP relay as their only read source. The ordinary related-event policy correctly rejects those untagged replaceable events, so discovery needs a narrow admission rule rather than a persistence-policy bypass. Maintain a shared derived set of authors owning roots in Full repositories and admit only kind 0 or kind 10002 from that set. Query both kinds in each bounded author batch, select only the NIP-01-preferred event per author and kind, and run them through ordinary write policy, persistence and broadcast. Re-read accepted relay lists from the local database before deriving inbox ownership, so a signed but rejected response cannot steer outbound connections. StateOnly roots remain inert and identity events cannot self-qualify an author. Rebuild inbox ownership from retained accepted relay lists on startup before any network refresh, then refresh that state remotely on the normal cadence. This keeps established conversation coverage independent of an index source. The accepted-root inventory remains authoritative and reconstructs removals on restart. This deliberately excludes response-author fanout, recursive scopes, general identity history, a rejected-identity cache and another deletion graph. Validation: - nix develop -c cargo check --lib - nix develop -c cargo test --lib identity_cannot_make_its_own_author_relevant - nix develop -c cargo test --lib state_only_root_becomes_an_inbox_source_after_promotion - nix develop -c cargo test --lib identity_selection_keeps_latest_profile_and_relay_list_per_author - nix develop -c cargo test --test sync root_author_inbox_reuses_existing_root_sync_pipeline -- --nocapture
1135 lines
46 KiB
Rust
1135 lines
46 KiB
Rust
/// Nostr Relay Builder Configuration
|
||
///
|
||
/// Wires nostr-relay-builder with NIP-34 admission/routing policy.
|
||
/// Deletion behaviour is owned by the `nostr::lifecycle` subsystem;
|
||
/// this module keeps top-level `admit_event`
|
||
/// dispatch and non-deletion admission/router policy logic.
|
||
use std::future::Future;
|
||
use std::net::SocketAddr;
|
||
use std::num::NonZeroUsize;
|
||
use std::path::Path;
|
||
use std::pin::Pin;
|
||
use std::sync::{Arc, RwLock};
|
||
use std::time::Duration;
|
||
|
||
use anyhow::Result;
|
||
use nostr::nips::nip19::ToBech32;
|
||
use nostr_lmdb::NostrLmdb;
|
||
use nostr_memory::MemoryDatabase;
|
||
use nostr_sdk::prelude::*;
|
||
|
||
use crate::config::{Config, DatabaseBackend};
|
||
use crate::nostr::events::RepositoryAnnouncement;
|
||
use crate::nostr::lifecycle::{
|
||
DeletionContext, DeletionRuntime, DeletionService, HoldingStore, ReplaceableHistoryStore,
|
||
RepositoryLifecycle, Tombstones,
|
||
};
|
||
use crate::nostr::persistence::{EventPersistence, SaveContext};
|
||
use crate::nostr::policy::{
|
||
accepted_purgatory, duplicate, reject_error, reject_invalid, reject_restricted,
|
||
AnnouncementPolicy, AnnouncementResult, IdentityAdmission, PolicyContext, PrEventPolicy,
|
||
ReferenceResult, RelatedEventPolicy, SharedProactiveRootAuthorIndex, StatePolicy, StateResult,
|
||
};
|
||
use crate::nostr::SharedDatabase;
|
||
use crate::purgatory::promotion_hooks::NostrPurgatoryPromotionHooks;
|
||
use crate::sync::rejected_index::RejectedEventsIndex;
|
||
|
||
/// Connection-wide frame allowance layered above the operation-specific
|
||
/// write, query, and authentication quotas.
|
||
///
|
||
/// rust-nostr's 300/minute default closed production browser clients during
|
||
/// ordinary bursty subscription churn. One hundred frames per second preserves a
|
||
/// bounded catch-all for malformed/non-operation traffic while leaving the
|
||
/// tighter operation quotas in charge of valid protocol work.
|
||
const CLIENT_MESSAGES_PER_MINUTE: u32 = 6_000;
|
||
|
||
/// Temporary compatibility override for rust-nostr 0.45's newly introduced
|
||
/// 120-query-per-minute default.
|
||
///
|
||
/// A finite per-connection bound remains useful as one DoS layer, but 120 is
|
||
/// too restrictive for legitimate sync and NIP-77 currently charges every
|
||
/// SDK-managed `NEG-MSG` continuation against the same allowance. Re-evaluate
|
||
/// this 10× value after upstream separates or otherwise revises that accounting.
|
||
const CLIENT_QUERIES_PER_MINUTE: u32 = 1_200;
|
||
|
||
/// Cumulative serialized REQ state retained for one client connection.
|
||
///
|
||
/// Repository sync legitimately keeps multiple byte-budgeted live filters
|
||
/// open at once. Production's 34-filter coverage reached about 1.2 MiB and
|
||
/// was partially rejected by rust-nostr's 1 MiB default. Five MiB remains a
|
||
/// finite per-connection allocation boundary while matching the largest
|
||
/// individual WebSocket message we already admit and leaving roughly 4x room
|
||
/// above the observed working set.
|
||
const MAX_SUBSCRIPTION_STATE_BYTES: usize = 5 * 1024 * 1024;
|
||
|
||
/// NIP-34 Write Policy — admission and routing for GRASP-01 events
|
||
///
|
||
/// Acts as the top-level admission gate and router. Each incoming event is:
|
||
/// 1. Checked against the event blacklist.
|
||
/// 2. Passed through the deletion gate (`DeletionService::gate`).
|
||
/// 3. Dispatched to the appropriate sub-policy:
|
||
/// - `AnnouncementPolicy` — repository announcement validation
|
||
/// - `StatePolicy` — state event validation + ref alignment
|
||
/// - `PrEventPolicy` — PR / PR-Update validation
|
||
/// - `RelatedEventPolicy` — forward/backward reference checking
|
||
/// - `DeletionService` — NIP-09 deletion and NIP-62 vanish handling
|
||
///
|
||
/// Deletion logic (cascade, holding, archive, recovery, startup passes) lives
|
||
/// entirely in `crate::nostr::lifecycle`. This struct exposes a `deletion()`
|
||
/// accessor for callers that need to reach those operations directly.
|
||
#[derive(Clone)]
|
||
pub struct Nip34WritePolicy {
|
||
ctx: PolicyContext,
|
||
announcement_policy: AnnouncementPolicy,
|
||
state_policy: StatePolicy,
|
||
pr_event_policy: PrEventPolicy,
|
||
related_event_policy: RelatedEventPolicy,
|
||
proactive_root_authors: SharedProactiveRootAuthorIndex,
|
||
deletion: DeletionService,
|
||
rejected_events_index: Arc<RwLock<Option<Arc<RejectedEventsIndex>>>>,
|
||
}
|
||
|
||
impl std::fmt::Debug for Nip34WritePolicy {
|
||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||
f.debug_struct("Nip34WritePolicy")
|
||
.field("domain", &self.ctx.domain)
|
||
.field("git_data_path", &self.ctx.git_data_path)
|
||
.field("database", &"<database>")
|
||
.finish()
|
||
}
|
||
}
|
||
|
||
impl Nip34WritePolicy {
|
||
#[allow(clippy::too_many_arguments)]
|
||
pub fn new(
|
||
database: SharedDatabase,
|
||
tombstones: Tombstones,
|
||
holding: HoldingStore,
|
||
lifecycle: RepositoryLifecycle,
|
||
history: ReplaceableHistoryStore,
|
||
git_data_path: impl Into<std::path::PathBuf>,
|
||
purgatory: std::sync::Arc<crate::purgatory::Purgatory>,
|
||
config: crate::config::Config,
|
||
repo_init_locks: crate::grasp06::receive::RepoInitLocks,
|
||
) -> Self {
|
||
let git_data_path = git_data_path.into();
|
||
let domain = config.domain.clone();
|
||
let deletion_ctx = DeletionContext::new(
|
||
domain.clone(),
|
||
database.clone(),
|
||
tombstones.clone(),
|
||
holding.clone(),
|
||
lifecycle,
|
||
history.clone(),
|
||
git_data_path.clone(),
|
||
purgatory.clone(),
|
||
config.clone(),
|
||
repo_init_locks.clone(),
|
||
);
|
||
|
||
let ctx = PolicyContext::new(domain, database, git_data_path, purgatory, config.clone());
|
||
Self {
|
||
announcement_policy: AnnouncementPolicy::new(ctx.clone(), config.clone()),
|
||
state_policy: StatePolicy::new(ctx.clone()),
|
||
pr_event_policy: PrEventPolicy::new(ctx.clone(), repo_init_locks),
|
||
related_event_policy: RelatedEventPolicy::new(ctx.clone()),
|
||
proactive_root_authors: crate::nostr::policy::ProactiveRootAuthorIndex::shared(),
|
||
deletion: DeletionService::new(deletion_ctx),
|
||
rejected_events_index: Arc::new(RwLock::new(None)),
|
||
ctx,
|
||
}
|
||
}
|
||
|
||
pub fn proactive_root_authors(&self) -> SharedProactiveRootAuthorIndex {
|
||
self.proactive_root_authors.clone()
|
||
}
|
||
|
||
async fn handle_proactive_identity(&self, event: &Event) -> WritePolicyResult {
|
||
match self.proactive_root_authors.admit(event).await {
|
||
IdentityAdmission::Accept => WritePolicyResult::Accept,
|
||
IdentityAdmission::IrrelevantAuthor => {
|
||
reject_restricted("Kind 0/10002 author must own an accepted repository root")
|
||
}
|
||
IdentityAdmission::NotIdentity => {
|
||
reject_invalid("Proactive identity policy received a non-identity event")
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Check if an event author is blacklisted
|
||
///
|
||
/// Returns Some(reason) if blacklisted, None if not blacklisted.
|
||
fn check_event_blacklist(&self, event: &Event) -> Option<String> {
|
||
let event_blacklist = self.ctx.config.event_blacklist_config();
|
||
if !event_blacklist.enabled() {
|
||
return None;
|
||
}
|
||
|
||
let npub = event.pubkey.to_bech32().ok()?;
|
||
event_blacklist.check(&npub)
|
||
}
|
||
|
||
/// Get a reference to the purgatory for read-only access
|
||
pub fn purgatory(&self) -> &std::sync::Arc<crate::purgatory::Purgatory> {
|
||
&self.ctx.purgatory
|
||
}
|
||
|
||
/// Get a reference to the holding store for retention cleanup orchestration.
|
||
pub fn holding(&self) -> &HoldingStore {
|
||
self.deletion.holding()
|
||
}
|
||
|
||
/// Get a reference to the replaceable-history store.
|
||
pub fn history(&self) -> &ReplaceableHistoryStore {
|
||
self.deletion.history()
|
||
}
|
||
|
||
pub fn tombstones(&self) -> &Tombstones {
|
||
self.deletion.tombstones()
|
||
}
|
||
|
||
/// Set the local relay for purgatory notifications.
|
||
///
|
||
/// This must be called after the relay is created since the relay depends
|
||
/// on this policy, but purgatory sync needs the relay to notify subscribers.
|
||
pub fn set_local_relay(&self, relay: nostr_sdk::local_relay::LocalRelay) {
|
||
self.ctx.set_local_relay(relay);
|
||
}
|
||
|
||
/// Make sync's rejected-event dependencies available to direct-write
|
||
/// purgatory promotions.
|
||
pub fn set_rejected_events_index(&self, index: Arc<RejectedEventsIndex>) {
|
||
*self
|
||
.rejected_events_index
|
||
.write()
|
||
.expect("rejected events index lock poisoned") = Some(index);
|
||
}
|
||
|
||
pub(crate) fn rejected_events_index(&self) -> Option<Arc<RejectedEventsIndex>> {
|
||
self.rejected_events_index
|
||
.read()
|
||
.expect("rejected events index lock poisoned")
|
||
.clone()
|
||
}
|
||
|
||
pub(crate) fn local_relay(&self) -> Option<nostr_sdk::local_relay::LocalRelay> {
|
||
self.ctx.get_local_relay()
|
||
}
|
||
|
||
/// Access the deletion subsystem directly.
|
||
///
|
||
/// Exposes startup passes and recovery without requiring delegation wrappers
|
||
/// on this struct. Repository lifecycle coordination is exposed separately
|
||
/// through [`RepositoryLifecycle`].
|
||
pub fn deletion(&self) -> &DeletionService {
|
||
&self.deletion
|
||
}
|
||
|
||
fn event_persistence(&self) -> EventPersistence<'_> {
|
||
EventPersistence::new(&self.ctx.database)
|
||
}
|
||
|
||
/// Save an already-admitted event through the central accepted-event path.
|
||
pub async fn save_accepted_event(
|
||
&self,
|
||
event: &Event,
|
||
context: SaveContext,
|
||
) -> anyhow::Result<()> {
|
||
self.event_persistence()
|
||
.save_accepted_event(event, context)
|
||
.await
|
||
}
|
||
|
||
/// Extract repository identifier from event's 'd' tag.
|
||
///
|
||
/// Used for structured logging when parsing fails - we try to extract
|
||
/// the identifier even if full parsing failed.
|
||
fn extract_identifier_from_event(event: &Event) -> String {
|
||
event
|
||
.tags
|
||
.iter()
|
||
.find(|t| t.kind() == "d")
|
||
.and_then(|t| t.content())
|
||
.map(|s| s.to_string())
|
||
.unwrap_or_else(|| "unknown".to_string())
|
||
}
|
||
|
||
/// Extract ALL repository identifiers from PR event's 'a' tags.
|
||
///
|
||
/// PR events can reference multiple repositories via multiple 'a' tags
|
||
/// (e.g., when there are multiple maintainers). Each tag has format
|
||
/// `30617:<owner_pubkey>:<identifier>`.
|
||
///
|
||
/// Returns a vector of unique identifiers, or `["unknown"]` if none found.
|
||
fn extract_repos_from_pr_event(event: &Event) -> Vec<String> {
|
||
let repos: Vec<String> = event
|
||
.tags
|
||
.iter()
|
||
.filter_map(|tag| {
|
||
let tag_vec = tag.clone().to_vec();
|
||
if tag_vec.len() >= 2 && tag_vec[0] == "a" && tag_vec[1].starts_with("30617:") {
|
||
// Format: 30617:<owner_pubkey>:<identifier>
|
||
let parts: Vec<&str> = tag_vec[1].split(':').collect();
|
||
if parts.len() >= 3 {
|
||
Some(parts[2].to_string())
|
||
} else {
|
||
None
|
||
}
|
||
} else {
|
||
None
|
||
}
|
||
})
|
||
.collect();
|
||
|
||
// Deduplicate while preserving order
|
||
let mut seen = std::collections::HashSet::new();
|
||
let unique_repos: Vec<String> = repos
|
||
.into_iter()
|
||
.filter(|r| seen.insert(r.clone()))
|
||
.collect();
|
||
|
||
if unique_repos.is_empty() {
|
||
vec!["unknown".to_string()]
|
||
} else {
|
||
unique_repos
|
||
}
|
||
}
|
||
|
||
/// Handle repository announcement event
|
||
async fn handle_announcement(&self, event: &Event) -> WritePolicyResult {
|
||
let event_id_str = event.id.to_bech32().unwrap_or_else(|_| event.id.to_hex());
|
||
|
||
let admission = self.announcement_policy.validate(event).await;
|
||
let deletion_hooks = self.deletion.admission_hooks();
|
||
let deletion_hook_result = deletion_hooks
|
||
.on_announcement_admission(event, &admission)
|
||
.await;
|
||
|
||
match admission {
|
||
AnnouncementResult::Accept | AnnouncementResult::AcceptArchive => {
|
||
// Parse announcement to get repository details
|
||
match RepositoryAnnouncement::from_event(event.clone()) {
|
||
Ok(announcement) => {
|
||
// Try to create bare repository if it doesn't exist
|
||
if let Err(e) = self
|
||
.announcement_policy
|
||
.ensure_bare_repository(&announcement)
|
||
{
|
||
tracing::warn!(
|
||
"Failed to create bare repository for {}: {}",
|
||
event_id_str,
|
||
e
|
||
);
|
||
// Note: We still accept the event even if repo creation fails
|
||
}
|
||
|
||
tracing::debug!("Accepted repository announcement: {}", event_id_str);
|
||
|
||
// Check purgatory for state events that might now be authorized
|
||
self.check_purgatory_state_events_for_identifier(&announcement.identifier)
|
||
.await;
|
||
|
||
WritePolicyResult::Accept
|
||
}
|
||
Err(e) => {
|
||
let npub = event
|
||
.pubkey
|
||
.to_bech32()
|
||
.unwrap_or_else(|_| event.pubkey.to_hex());
|
||
let event_id_short = &event.id.to_hex()[..12];
|
||
// Try to extract repo identifier from 'd' tag even if parsing failed
|
||
let repo = Self::extract_identifier_from_event(event);
|
||
// Structured log for migration scripts
|
||
tracing::warn!(
|
||
"[PARSE_FAIL] kind={} event_id={}... reason=\"{}\" repo={} npub={}",
|
||
event.kind.as_u16(),
|
||
event_id_short,
|
||
e,
|
||
repo,
|
||
npub
|
||
);
|
||
reject_invalid(format!("Failed to parse announcement: {}", e))
|
||
}
|
||
}
|
||
}
|
||
AnnouncementResult::AcceptPurgatory => {
|
||
let announcement = match RepositoryAnnouncement::from_event(event.clone()) {
|
||
Ok(announcement) => announcement,
|
||
Err(e) => {
|
||
return reject_invalid(format!("Failed to parse announcement: {}", e))
|
||
}
|
||
};
|
||
|
||
if deletion_hook_result.recovered_repository() {
|
||
tracing::info!(
|
||
identifier = %announcement.identifier,
|
||
event_id = %event_id_str,
|
||
"Accepted re-announcement with recovery"
|
||
);
|
||
if let Err(e) = self
|
||
.announcement_policy
|
||
.ensure_bare_repository(&announcement)
|
||
{
|
||
tracing::warn!(
|
||
"Failed to ensure bare repository after recovery for {}: {}",
|
||
event_id_str,
|
||
e
|
||
);
|
||
}
|
||
return WritePolicyResult::Accept;
|
||
}
|
||
|
||
// New announcement - add to purgatory
|
||
match self.announcement_policy.add_to_purgatory(event) {
|
||
Ok(()) => {
|
||
// The server may already have the repository's owner-authored state
|
||
// and Git objects. Reapply that stored state now so a newly accepted
|
||
// maintainer announcement can be promoted without waiting for the
|
||
// same replaceable state event to arrive over sync again.
|
||
self.reconcile_stored_state_events_for_identifier(&announcement.identifier)
|
||
.await;
|
||
|
||
tracing::info!(
|
||
"Accepted announcement to purgatory: {} (waiting for git data)",
|
||
event_id_str
|
||
);
|
||
accepted_purgatory("won't be served until git data arrives")
|
||
}
|
||
Err(e) => {
|
||
tracing::warn!(
|
||
"Failed to add announcement to purgatory {}: {}",
|
||
event_id_str,
|
||
e
|
||
);
|
||
reject_error(e)
|
||
}
|
||
}
|
||
}
|
||
AnnouncementResult::AcceptMaintainer => {
|
||
// Parse announcement to get details for logging
|
||
match RepositoryAnnouncement::from_event(event.clone()) {
|
||
Ok(announcement) => {
|
||
tracing::info!(
|
||
"Accepted maintainer announcement {} (author {} is listed as maintainer for {})",
|
||
event_id_str,
|
||
event.pubkey.to_hex(),
|
||
announcement.identifier
|
||
);
|
||
// Don't create bare repository for external announcements
|
||
|
||
// Check purgatory for state events that might now be authorized
|
||
self.check_purgatory_state_events_for_identifier(&announcement.identifier)
|
||
.await;
|
||
|
||
WritePolicyResult::Accept
|
||
}
|
||
Err(e) => {
|
||
let npub = event
|
||
.pubkey
|
||
.to_bech32()
|
||
.unwrap_or_else(|_| event.pubkey.to_hex());
|
||
let event_id_short = &event.id.to_hex()[..12];
|
||
// Try to extract repo identifier from 'd' tag even if parsing failed
|
||
let repo = Self::extract_identifier_from_event(event);
|
||
// Structured log for migration scripts
|
||
tracing::warn!(
|
||
"[PARSE_FAIL] kind={} event_id={}... reason=\"{}\" repo={} npub={}",
|
||
event.kind.as_u16(),
|
||
event_id_short,
|
||
e,
|
||
repo,
|
||
npub
|
||
);
|
||
reject_invalid(format!("Failed to parse announcement: {}", e))
|
||
}
|
||
}
|
||
}
|
||
AnnouncementResult::Reject(reason) => {
|
||
tracing::warn!(
|
||
"Rejected repository announcement {}: {}",
|
||
event_id_str,
|
||
reason
|
||
);
|
||
reject_invalid(reason)
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Handle repository state event
|
||
///
|
||
/// # Arguments
|
||
/// * `event` - The state event to validate
|
||
/// * `is_synced` - True if this event came from proactive sync (vs user-submitted)
|
||
async fn handle_state(&self, event: &Event, is_synced: bool) -> WritePolicyResult {
|
||
match self.state_policy.validate(event) {
|
||
StateResult::Accept => {
|
||
let promotion_hooks = NostrPurgatoryPromotionHooks::recovery_only(self);
|
||
|
||
// Process state alignment asynchronously
|
||
match self
|
||
.state_policy
|
||
.process_state_event(event, is_synced, Some(&promotion_hooks))
|
||
.await
|
||
{
|
||
Ok(policy_result) => {
|
||
self.deletion
|
||
.admission_hooks()
|
||
.on_state_admission(event, &policy_result)
|
||
.await;
|
||
policy_result
|
||
}
|
||
Err(e) => {
|
||
let npub = event
|
||
.pubkey
|
||
.to_bech32()
|
||
.unwrap_or_else(|_| event.pubkey.to_hex());
|
||
let event_id_short = &event.id.to_hex()[..12];
|
||
// Try to extract repo identifier from 'd' tag even if parsing failed
|
||
let repo = Self::extract_identifier_from_event(event);
|
||
// Structured log for migration scripts
|
||
tracing::warn!(
|
||
"[PARSE_FAIL] kind={} event_id={}... reason=\"{}\" repo={} npub={}",
|
||
event.kind.as_u16(),
|
||
event_id_short,
|
||
e,
|
||
repo,
|
||
npub
|
||
);
|
||
// reject if processing failed
|
||
reject_error(format!("{e}"))
|
||
}
|
||
}
|
||
}
|
||
StateResult::Reject(reason) => {
|
||
let npub = event
|
||
.pubkey
|
||
.to_bech32()
|
||
.unwrap_or_else(|_| event.pubkey.to_hex());
|
||
let event_id_short = &event.id.to_hex()[..12];
|
||
// Try to extract repo identifier from 'd' tag even if parsing failed
|
||
let repo = Self::extract_identifier_from_event(event);
|
||
// Structured log for migration scripts
|
||
tracing::warn!(
|
||
"[PARSE_FAIL] kind={} event_id={}... reason=\"{}\" repo={} npub={}",
|
||
event.kind.as_u16(),
|
||
event_id_short,
|
||
reason,
|
||
repo,
|
||
npub
|
||
);
|
||
reject_invalid(reason)
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Handle PR or PR Update event
|
||
///
|
||
/// # Arguments
|
||
/// * `event` - The PR event to validate
|
||
/// * `is_synced` - True if this event came from proactive sync (vs user-submitted)
|
||
async fn handle_pr_event(&self, event: &Event, is_synced: bool) -> WritePolicyResult {
|
||
let event_id_str = event.id.to_bech32().unwrap_or_else(|_| event.id.to_hex());
|
||
|
||
// duplicate check in purgatory
|
||
let in_purgatory = self
|
||
.ctx
|
||
.purgatory
|
||
.find_pr(&event.id.to_hex())
|
||
.is_some_and(|e| e.event.is_some());
|
||
if in_purgatory {
|
||
tracing::debug!(
|
||
"processed PR event duplicate (already in purgatory): {}",
|
||
event.id,
|
||
);
|
||
return duplicate("in purgatory");
|
||
}
|
||
|
||
// duplicate check in db
|
||
//
|
||
// Note: the database runs with `process_nip09(false)`, so `check_id`
|
||
// only ever returns `Saved` / `NotExistent` here — re-submission of a
|
||
// deleted PR is already rejected upstream by `DeletionService::gate`.
|
||
match &self.ctx.database.check_id(&event.id).await {
|
||
Ok(DatabaseEventStatus::Saved) => {
|
||
return duplicate("already have this event");
|
||
}
|
||
Err(e) => {
|
||
return reject_error(format!("internal error: {e}"));
|
||
}
|
||
_ => {} // continue
|
||
}
|
||
|
||
// Reject PRs unrelated to stored repositories / events.
|
||
//
|
||
// GRASP-06 (06.md lines 19–22) relaxes this: a PR / PR-Update event
|
||
// whose `clone` tag names this relay's
|
||
// `/prs/<signer-npub>/<d>.git` endpoint (and which carries a
|
||
// matching `a` tag of the form `30617:<hex>:<d>`) MUST be accepted
|
||
// even when no announcement for the target coord is on this relay.
|
||
// Such events skip the related-event check and fall through to the
|
||
// git-data-check + purgatory branch below.
|
||
match self.handle_related_event(event, "PR").await {
|
||
WritePolicyResult::Accept => {} // continue
|
||
rejected => {
|
||
if !crate::grasp06::policy::event_qualifies_for_pr_relaxation(
|
||
event,
|
||
&self.ctx.config,
|
||
) {
|
||
return rejected;
|
||
}
|
||
tracing::info!(
|
||
"Accepted orphan PR event {} under GRASP-06: clone tag names this relay's /prs/ endpoint",
|
||
event_id_str,
|
||
);
|
||
}
|
||
}
|
||
|
||
// Check if git data exists (delete any incorrect commits at refs/nostr/<event-id>, copies correct data to relivant repositories)
|
||
match self.pr_event_policy.git_data_check(event).await {
|
||
Ok(false) => {
|
||
// Only reject expired events if they're from sync (not user-submitted)
|
||
// User-submitted events should be allowed to retry in case git data became available
|
||
if is_synced && self.ctx.purgatory.is_expired(&event.id) {
|
||
tracing::debug!(
|
||
event_id = %event_id_str,
|
||
"PR event previously expired from purgatory (synced), rejecting to prevent re-sync loop"
|
||
);
|
||
return reject_invalid("previously expired from purgatory without git data");
|
||
}
|
||
|
||
// No git data exists - add to purgatory
|
||
let commit = event
|
||
.tags
|
||
.iter()
|
||
.find_map(|tag| {
|
||
let tag_vec = tag.clone().to_vec();
|
||
if tag_vec.len() >= 2 && tag_vec[0] == "c" {
|
||
Some(tag_vec[1].clone())
|
||
} else {
|
||
None
|
||
}
|
||
})
|
||
.unwrap_or_else(|| "unknown".to_string());
|
||
|
||
tracing::info!(
|
||
"PR event {} added to purgatory: waiting for git push with commit {}",
|
||
event_id_str,
|
||
commit
|
||
);
|
||
|
||
// Add to purgatory
|
||
self.ctx.purgatory.add_pr(
|
||
event.clone(),
|
||
event.id.to_hex(),
|
||
commit.clone(),
|
||
is_synced,
|
||
);
|
||
|
||
accepted_purgatory(format!(
|
||
"PR event stored, waiting for git push with commit {}",
|
||
commit
|
||
))
|
||
}
|
||
Ok(true) => {
|
||
// Git data exists - proceed with normal validation
|
||
tracing::debug!("Git data exists for PR event {}", event_id_str);
|
||
WritePolicyResult::Accept
|
||
}
|
||
Err(e) => {
|
||
// Error checking git data - reject event
|
||
let npub = event
|
||
.pubkey
|
||
.to_bech32()
|
||
.unwrap_or_else(|_| event.pubkey.to_hex());
|
||
let event_id_short = &event.id.to_hex()[..12];
|
||
// Extract ALL repo identifiers from 'a' tags for PR events
|
||
// (PR events can reference multiple repos when there are multiple maintainers)
|
||
let repos = Self::extract_repos_from_pr_event(event);
|
||
// Structured log for migration scripts - log once per repo
|
||
for repo in &repos {
|
||
tracing::warn!(
|
||
"[PARSE_FAIL] kind={} event_id={}... reason=\"git data check failed: {}\" repo={} npub={}",
|
||
event.kind.as_u16(),
|
||
event_id_short,
|
||
e,
|
||
repo,
|
||
npub
|
||
);
|
||
}
|
||
reject_error(format!("Failed to check git data: {}", e))
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Check purgatory for state events that might now be authorized by a new announcement
|
||
///
|
||
/// When an announcement is accepted, state events in purgatory that were previously
|
||
/// rejected due to missing announcements might now be authorized. This method:
|
||
/// 1. Finds all state events in purgatory for the identifier
|
||
/// 2. Re-evaluates authorization for each event
|
||
/// 3. Processes authorized events (releases from purgatory)
|
||
/// 4. Keeps unauthorized events in purgatory (will expire naturally)
|
||
async fn check_purgatory_state_events_for_identifier(&self, identifier: &str) {
|
||
let state_events = self.ctx.purgatory.find_state(identifier);
|
||
|
||
if state_events.is_empty() {
|
||
return;
|
||
}
|
||
|
||
tracing::debug!(
|
||
identifier = %identifier,
|
||
count = state_events.len(),
|
||
"Checking purgatory state events after announcement acceptance"
|
||
);
|
||
|
||
for entry in state_events {
|
||
let promotion_hooks = NostrPurgatoryPromotionHooks::recovery_only(self);
|
||
|
||
// Re-evaluate authorization with the new announcement
|
||
match self
|
||
.state_policy
|
||
.process_state_event(&entry.event, false, Some(&promotion_hooks))
|
||
.await
|
||
{
|
||
Ok(WritePolicyResult::Accept) => {
|
||
tracing::info!(
|
||
event_id = %entry.event.id,
|
||
identifier = %identifier,
|
||
"State event in purgatory now authorized, will be processed"
|
||
);
|
||
// Event will be automatically removed from purgatory by process_state_event
|
||
// and broadcast to subscribers
|
||
}
|
||
Ok(WritePolicyResult::Reject { message, .. }) => {
|
||
if message.contains("not authorized") {
|
||
tracing::debug!(
|
||
event_id = %entry.event.id,
|
||
identifier = %identifier,
|
||
"State event in purgatory still not authorized, keeping in purgatory"
|
||
);
|
||
// Keep in purgatory - will expire naturally after 30 minutes
|
||
} else {
|
||
tracing::debug!(
|
||
event_id = %entry.event.id,
|
||
identifier = %identifier,
|
||
reason = %message,
|
||
"State event in purgatory rejected for other reason"
|
||
);
|
||
}
|
||
}
|
||
Err(e) => {
|
||
tracing::warn!(
|
||
event_id = %entry.event.id,
|
||
identifier = %identifier,
|
||
error = %e,
|
||
"Error re-evaluating state event in purgatory"
|
||
);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Reapply stored state after a new maintainer repository enters purgatory.
|
||
///
|
||
/// A shared GRASP server can already hold the owner's latest state and Git
|
||
/// objects when an invitee publishes their reciprocal announcement. Since
|
||
/// the state is replaceable and already stored, sync may not deliver it
|
||
/// again. Processing stored candidates newest-first lets each maintainer
|
||
/// chain apply its preferred state before considering older candidates,
|
||
/// without requiring an invitee-authored state event or push.
|
||
async fn reconcile_stored_state_events_for_identifier(&self, identifier: &str) {
|
||
let filter = Filter::new()
|
||
.kind(Kind::RepoState)
|
||
.custom_tag(SingleLetterTag::LOWERCASE_D, identifier.to_string());
|
||
let mut states: Vec<Event> = match self.ctx.database.query(filter).await {
|
||
Ok(events) => events.into_iter().collect(),
|
||
Err(error) => {
|
||
tracing::warn!(
|
||
identifier = %identifier,
|
||
error = %error,
|
||
"Failed to query stored state for new maintainer repository"
|
||
);
|
||
return;
|
||
}
|
||
};
|
||
|
||
states.sort_by(|left, right| {
|
||
right
|
||
.created_at
|
||
.cmp(&left.created_at)
|
||
.then_with(|| left.id.cmp(&right.id))
|
||
});
|
||
|
||
for state in states {
|
||
let promotion_hooks = NostrPurgatoryPromotionHooks::recovery_only(self);
|
||
if let Err(error) = self
|
||
.state_policy
|
||
.process_state_event(&state, true, Some(&promotion_hooks))
|
||
.await
|
||
{
|
||
tracing::warn!(
|
||
identifier = %identifier,
|
||
event_id = %state.id,
|
||
error = %error,
|
||
"Failed to reapply stored state to new maintainer repository"
|
||
);
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Handle events that must reference accepted repositories or events
|
||
async fn handle_related_event(&self, event: &Event, event_type: &str) -> WritePolicyResult {
|
||
let event_id_str = event.id.to_bech32().unwrap_or_else(|_| event.id.to_hex());
|
||
|
||
match self.related_event_policy.check_references(event).await {
|
||
Ok(ReferenceResult::ReferencesRepository(addr_ref)) => {
|
||
tracing::debug!(
|
||
"Accepted {} event {}: references accepted repository {}",
|
||
event_type,
|
||
event_id_str,
|
||
addr_ref
|
||
);
|
||
WritePolicyResult::Accept
|
||
}
|
||
Ok(ReferenceResult::ReferencesEvent(event_ref)) => {
|
||
tracing::debug!(
|
||
"Accepted {} event {}: references accepted event {}",
|
||
event_type,
|
||
event_id_str,
|
||
event_ref
|
||
);
|
||
WritePolicyResult::Accept
|
||
}
|
||
Ok(ReferenceResult::ReferencedByAccepted) => {
|
||
tracing::debug!(
|
||
"Accepted {} event {}: referenced by accepted event",
|
||
event_type,
|
||
event_id_str
|
||
);
|
||
WritePolicyResult::Accept
|
||
}
|
||
Ok(ReferenceResult::Orphan) => {
|
||
let (addressable_refs, event_refs) =
|
||
RelatedEventPolicy::extract_reference_tags(event);
|
||
tracing::info!(
|
||
"Rejected orphan {} event {} (kind={}, pubkey={}): no references to accepted repos or events (checked {} addressable, {} event refs)",
|
||
event_type,
|
||
event.id.to_hex(),
|
||
event.kind.as_u16(),
|
||
event.pubkey.to_hex(),
|
||
addressable_refs.len(),
|
||
event_refs.len()
|
||
);
|
||
reject_restricted(format!(
|
||
"{} event must reference an accepted repository or accepted event",
|
||
event_type
|
||
))
|
||
}
|
||
Err(e) => {
|
||
tracing::warn!(
|
||
"Database query failed for {} {}, rejecting (fail-secure): {}",
|
||
event_type,
|
||
event_id_str,
|
||
e
|
||
);
|
||
reject_error(format!("Database query failed: {}", e))
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
impl WritePolicy for Nip34WritePolicy {
|
||
fn admit_event<'a>(
|
||
&'a self,
|
||
event: &'a nostr_sdk::prelude::Event,
|
||
addr: &'a SocketAddr,
|
||
) -> Pin<Box<dyn Future<Output = WritePolicyResult> + Send + 'a>> {
|
||
Box::pin(async move {
|
||
// Check event blacklist FIRST - it overrides everything
|
||
if let Some(reason) = self.check_event_blacklist(event) {
|
||
tracing::debug!(
|
||
event_id = %event.id.to_bech32().unwrap_or_else(|_| event.id.to_hex()),
|
||
author = %event.pubkey.to_hex(),
|
||
reason = %reason,
|
||
"Rejected event from blacklisted author"
|
||
);
|
||
return WritePolicyResult::reject(MachineReadablePrefix::Blocked, reason);
|
||
}
|
||
|
||
// Deletion / vanish gate.
|
||
//
|
||
// The relay-builder library has a `check_id` gate that runs *before*
|
||
// this policy and used to reject re-submitted deleted events with
|
||
// "this event is deleted" — but that gate is fed by the LMDB
|
||
// backend's internal NIP-09/NIP-62 tables, which go silent now that
|
||
// we run the backend with `process_nip09(false)` /
|
||
// `process_nip62(false)`. We reproduce that rejection here from our
|
||
// own persistent `Tombstones` store so deleted events (and events
|
||
// from vanished pubkeys) stay rejected across restarts.
|
||
if let Some(rejection) = self.deletion.gate(event).await {
|
||
return rejection;
|
||
}
|
||
|
||
// Detect if this is a synced event (from proactive sync) vs user-submitted
|
||
// Sync uses localhost:0 as a dummy address
|
||
let is_synced = addr.ip().is_loopback() && addr.port() == 0;
|
||
|
||
let result = match event.kind {
|
||
Kind::Metadata | Kind::RelayList => self.handle_proactive_identity(event).await,
|
||
Kind::GitRepoAnnouncement => self.handle_announcement(event).await,
|
||
Kind::RepoState => self.handle_state(event, is_synced).await,
|
||
Kind::GitPullRequest | Kind::GitPullRequestUpdate => {
|
||
self.handle_pr_event(event, is_synced).await
|
||
}
|
||
Kind::GitUserGraspList => {
|
||
// Accept all kind 10317 (User Grasp List) events
|
||
// for better GRASP repository discovery
|
||
tracing::debug!(
|
||
event_id = %event.id.to_bech32().unwrap_or_else(|_| event.id.to_hex()),
|
||
author = %event.pubkey.to_hex(),
|
||
"Accepted kind 10317 user grasp list"
|
||
);
|
||
WritePolicyResult::Accept
|
||
}
|
||
Kind::EventDeletion => self.deletion.handle_nip09(event).await,
|
||
Kind::RequestToVanish => self.deletion.handle_vanish(event).await,
|
||
_ => self.handle_related_event(event, "Event").await,
|
||
};
|
||
|
||
result
|
||
})
|
||
}
|
||
}
|
||
|
||
/// Storage handles shared by the relay runtime and HTTP/runtime services.
|
||
pub struct RelayStores {
|
||
/// Main relay database used for event queries and persistence.
|
||
pub database: SharedDatabase,
|
||
/// Persistent NIP-09/NIP-62 tombstone store.
|
||
pub tombstones: Tombstones,
|
||
/// Holding store used for deletion archival + expiry cleanup
|
||
pub holding: HoldingStore,
|
||
/// Replaceable-history store used for rollback history capture.
|
||
pub history: ReplaceableHistoryStore,
|
||
}
|
||
|
||
/// Runtime components produced when constructing the local relay.
|
||
pub struct RelayRuntime {
|
||
/// The local relay instance.
|
||
pub relay: LocalRelay,
|
||
/// The write policy used for event admission, routing, and persistence.
|
||
pub write_policy: Nip34WritePolicy,
|
||
/// Deletion service shared with write policy for NIP-09/NIP-62 operations.
|
||
pub deletion: DeletionService,
|
||
/// Runtime facade for deletion startup tasks and background cleanup.
|
||
pub deletion_runtime: DeletionRuntime,
|
||
/// Repository lifecycle locks shared by HTTP serving and deletion/recovery.
|
||
pub lifecycle: RepositoryLifecycle,
|
||
/// Storage handles used by the relay runtime.
|
||
pub stores: RelayStores,
|
||
}
|
||
|
||
/// Create a configured LocalRelay with full GRASP-01 validation
|
||
///
|
||
/// Returns a `RelayRuntime` struct containing the relay, write policy, deletion
|
||
/// runtime, lifecycle locks, and storage handles used by HTTP/runtime services.
|
||
pub async fn create_relay(
|
||
config: &Config,
|
||
purgatory: Arc<crate::purgatory::Purgatory>,
|
||
repo_init_locks: crate::grasp06::receive::RepoInitLocks,
|
||
) -> Result<RelayRuntime> {
|
||
tracing::info!("Configuring nostr relay with GRASP-01 validation...");
|
||
|
||
// Determine database path
|
||
let db_path = Path::new(&config.relay_data_path);
|
||
|
||
// Create database based on configuration
|
||
//
|
||
// NIP-09 (deletion) and NIP-62 (request to vanish) auto-processing is
|
||
// explicitly disabled on the main database: ngit-grasp owns this handling
|
||
// (see the deletion subsystem facade + `Tombstones` store).
|
||
// Leaving the backend defaults (`true`) would double-process: the backend
|
||
// would silently hard-delete events and block re-submission via its own
|
||
// internal tables that we cannot inspect, conflicting with our purgatory /
|
||
// bare-repo bookkeeping.
|
||
let (database, tombstones, holding, history): (
|
||
SharedDatabase,
|
||
Tombstones,
|
||
HoldingStore,
|
||
ReplaceableHistoryStore,
|
||
) = match config.database_backend {
|
||
DatabaseBackend::Memory => {
|
||
tracing::info!("Using in-memory database (no persistence)");
|
||
// Disable the backend's built-in NIP-09 / NIP-62 auto-processing to
|
||
// match the LMDB path: ngit-grasp owns this handling (deletion
|
||
// subsystem + Tombstones store). Leaving the memory
|
||
// backend's defaults (`true`) would silently suppress targeted events
|
||
// at query time the moment a kind-5 is *stored* — which in particular
|
||
// defeats `deletion_request_disrespector` (archival) mode, where the
|
||
// kind-5 is intentionally stored but not acted upon.
|
||
let db = MemoryDatabase::builder()
|
||
.max_events(NonZeroUsize::new(100_000).unwrap())
|
||
.process_nip09(false)
|
||
.process_nip62(false)
|
||
.build();
|
||
(
|
||
Arc::new(db),
|
||
Tombstones::in_memory(),
|
||
HoldingStore::in_memory(),
|
||
ReplaceableHistoryStore::in_memory(),
|
||
)
|
||
}
|
||
DatabaseBackend::Lmdb => {
|
||
tracing::info!("Using LMDB backend at: {}", db_path.display());
|
||
// Ensure the database directory exists
|
||
std::fs::create_dir_all(db_path).map_err(|e| {
|
||
anyhow::anyhow!(
|
||
"Failed to create LMDB directory {}: {}",
|
||
db_path.display(),
|
||
e
|
||
)
|
||
})?;
|
||
let db = NostrLmdb::builder(db_path)
|
||
.process_nip09(false)
|
||
.process_nip62(false)
|
||
.build()
|
||
.await
|
||
.map_err(|e| {
|
||
anyhow::anyhow!(
|
||
"Failed to open LMDB database at {}: {}",
|
||
db_path.display(),
|
||
e
|
||
)
|
||
})?;
|
||
let tombstones = Tombstones::open_lmdb(db_path).await?;
|
||
let holding =
|
||
HoldingStore::open_lmdb(db_path, Path::new(&config.effective_git_data_path()))
|
||
.await?;
|
||
let history = ReplaceableHistoryStore::open_lmdb(db_path).await?;
|
||
(Arc::new(db), tombstones, holding, history)
|
||
}
|
||
};
|
||
|
||
// Build relay with GRASP-01 validation
|
||
// Clone Arc for the write policy so both relay and policy can access the database
|
||
let git_data_path = config.effective_git_data_path();
|
||
|
||
// Log archive configuration (config.validate() must be called at startup)
|
||
let archive_config = config.archive_config();
|
||
if archive_config.enabled() {
|
||
tracing::info!(
|
||
"GRASP-05 archive mode enabled: archive_all={}, whitelist_entries={}, read_only={}",
|
||
archive_config.archive_all,
|
||
archive_config.whitelist.len(),
|
||
archive_config.read_only
|
||
);
|
||
}
|
||
|
||
// Log repository configuration
|
||
let repository_config = config.repository_config();
|
||
if repository_config.enabled() {
|
||
tracing::info!(
|
||
"Repository whitelist enabled: whitelist_entries={}",
|
||
repository_config.whitelist.len()
|
||
);
|
||
}
|
||
|
||
let lifecycle = match config.database_backend {
|
||
DatabaseBackend::Memory => RepositoryLifecycle::in_memory(),
|
||
DatabaseBackend::Lmdb => RepositoryLifecycle::for_git_data_path(Path::new(&git_data_path)),
|
||
};
|
||
|
||
// Create write policy with purgatory integration
|
||
let write_policy = Nip34WritePolicy::new(
|
||
database.clone(),
|
||
tombstones.clone(),
|
||
holding.clone(),
|
||
lifecycle.clone(),
|
||
history.clone(),
|
||
&git_data_path,
|
||
purgatory,
|
||
config.clone(),
|
||
repo_init_locks,
|
||
);
|
||
|
||
let mut builder = LocalRelayBuilder::default()
|
||
.database(database.clone())
|
||
.write_policy(write_policy.clone())
|
||
.rate_limit(RateLimit {
|
||
max_reqs: config.relay_max_subscriptions,
|
||
notes_per_minute: 60,
|
||
})
|
||
.queries_per_minute(CLIENT_QUERIES_PER_MINUTE)
|
||
.auth_events_per_minute(30)
|
||
.messages_per_minute(CLIENT_MESSAGES_PER_MINUTE)
|
||
.max_websocket_message_size(5 * 1024 * 1024)
|
||
.max_event_size(config.relay_max_event_size_bytes)
|
||
.websocket_handshake_timeout(Duration::from_secs(10))
|
||
.max_subid_length(250)
|
||
.max_filters_per_req(20)
|
||
.max_subscription_bytes(MAX_SUBSCRIPTION_STATE_BYTES)
|
||
.max_negentropy_subscriptions(10)
|
||
.max_negentropy_items(50_000)
|
||
.max_filter_limit(config.relay_filter_limit)
|
||
.max_query_results(config.relay_filter_limit)
|
||
.default_filter_limit(config.relay_filter_limit);
|
||
|
||
// `LocalRelayBuilder` otherwise applies rust-nostr's own finite default.
|
||
// In ngit-grasp, an unset limit deliberately delegates admission control to
|
||
// the OS and surrounding infrastructure, as documented by every config
|
||
// surface. Explicit operator limits remain exact.
|
||
builder = builder.max_connections(effective_max_connections(config.max_connections));
|
||
|
||
let relay = builder.build();
|
||
|
||
let deletion_runtime = DeletionRuntime::new(
|
||
write_policy.deletion().clone(),
|
||
holding.clone(),
|
||
config.holding_retention(),
|
||
config.holding_cleanup_interval(),
|
||
);
|
||
|
||
tracing::info!(
|
||
"Relay configured with GRASP-01 validation for domain: {}",
|
||
config.domain
|
||
);
|
||
|
||
let deletion = write_policy.deletion().clone();
|
||
|
||
Ok(RelayRuntime {
|
||
relay,
|
||
write_policy,
|
||
deletion,
|
||
deletion_runtime,
|
||
lifecycle,
|
||
stores: RelayStores {
|
||
database,
|
||
tombstones,
|
||
holding,
|
||
history,
|
||
},
|
||
})
|
||
}
|
||
|
||
fn effective_max_connections(configured: Option<usize>) -> usize {
|
||
configured.unwrap_or(tokio::sync::Semaphore::MAX_PERMITS)
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod connection_limit_tests {
|
||
use super::effective_max_connections;
|
||
|
||
#[test]
|
||
fn unset_connection_limit_overrides_dependency_default() {
|
||
assert_eq!(
|
||
effective_max_connections(None),
|
||
tokio::sync::Semaphore::MAX_PERMITS
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn configured_connection_limit_remains_exact() {
|
||
assert_eq!(effective_max_connections(Some(2)), 2);
|
||
}
|
||
}
|