pharos_native/runtime/

event_discovery.rs

//! Evidence-backed event ingestion for apps declaring an event-discovery contract.
//! Public sources are fetched only by the background worker, through the bounded transport.

use std::collections::BTreeMap;
use std::time::Duration;

use pharos_runtime_database::operations::{
    Mutation, Predicate, Query, SortKey, Transaction, WriteValue,
};
use serde_json::{Map, Value, json};

mod calendar_feed;
mod model;
mod social_announcements;
pub(super) mod social_evidence;

use super::{
    AppRuntime, app_database_operations, current_timestamp_utc, dynamic, runtime_timestamp_seconds,
};

type PreviewRegistry = BTreeMap<String, (AppRuntime, std::time::Instant)>;

const PROFILE_FIELDS: [&str; 8] = [
    "website",
    "instagram",
    "facebook",
    "source_url",
    "soundcloud",
    "spotify",
    "youtube",
    "x_url",
];

const SOURCE_CHECK_INTERVAL_SECONDS: u64 = 12 * 60 * 60;

/// Identifies providers that require an authorized connector instead of a public-page fetch.
fn requires_social_authorization(url: &url::Url) -> bool {
    pharos_runtime_social::provider::SocialProvider::from_url(url).is_some()
}

/// Recognizes common soft-login documents only when no usable event evidence was extracted.
fn password_gate(body: &str) -> bool {
    let lower = body.to_ascii_lowercase();
    lower.contains("type=\"password\"")
        || lower.contains("type='password'")
        || lower.contains("type=password")
        || ["<title>login", "<title>log in", "<title>sign in"]
            .iter()
            .any(|title| lower.contains(title))
}

/// Starts a managed discovery poller so external sources cannot delay ordinary queued jobs.
pub(super) fn start_worker(
    stack: std::sync::Arc<std::sync::RwLock<super::RuntimeStack>>,
) -> Option<super::job::JobWorkerState> {
    if !stack.read().ok()?.apps.iter().any(|app| {
        app.queue_enabled
            || app.app_id.as_deref() == Some("architect")
            || dynamic::dynamic_contract_doc(app, "social_connections_contract")
                .and_then(|value| value.get("enabled").and_then(Value::as_bool))
                == Some(true)
    }) {
        return None;
    }
    let state = super::job::JobWorkerState::default()
        .with_poll_interval_ms(5000)
        .with_poll_handler(std::sync::Arc::new(move || {
            let Some(apps) = stack.read().ok().map(|stack| stack.apps.clone()) else {
                return;
            };
            for app in &apps {
                if super::runtime_shutdown_requested() {
                    return;
                }
                #[cfg(not(target_os = "wasi"))]
                super::social_connections::process_one(app);
                process_one_source(app);
            }
            if !super::runtime_shutdown_requested() {
                process_registered_previews();
            }
        }));
    super::job::job_worker_start(&state, None).ok()?;
    Some(state)
}

/// Keeps bounded accepted preview runtimes available to the shared background worker.
fn preview_registry() -> &'static std::sync::Mutex<PreviewRegistry> {
    static REGISTRY: std::sync::OnceLock<std::sync::Mutex<PreviewRegistry>> =
        std::sync::OnceLock::new();
    REGISTRY.get_or_init(|| std::sync::Mutex::new(BTreeMap::new()))
}

/// Registers an explicitly enabled discovery contract in its existing isolated preview namespace.
pub(super) fn register_preview(app: &AppRuntime) {
    if !app.is_editor_preview {
        return;
    }
    let enabled = dynamic::dynamic_contract_doc(app, "event_discovery_contract")
        .and_then(|value| value.get("enabled").and_then(Value::as_bool))
        == Some(true)
        || dynamic::dynamic_contract_doc(app, "social_connections_contract")
            .and_then(|value| value.get("enabled").and_then(Value::as_bool))
            == Some(true);
    let (Some(config), Some(id)) = (&app.runtime_conf_path, &app.app_id) else {
        return;
    };
    let Ok(mut registry) = preview_registry().lock() else {
        return;
    };
    let now = std::time::Instant::now();
    registry.retain(|_, (_, seen)| now.duration_since(*seen) < Duration::from_secs(86400));
    let key = format!("{config}\0{id}");
    if !enabled {
        registry.remove(&key);
        return;
    }
    if registry.len() >= 32 && !registry.contains_key(&key) {
        return;
    }
    let mut preview = app.clone();
    preview.queue_enabled = true;
    registry.insert(key, (preview, now));
}

/// Processes one registered preview per poll with bounded round-robin scheduling.
pub(super) fn process_registered_previews() {
    static NEXT: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
    let preview = preview_registry().lock().ok().and_then(|registry| {
        if registry.is_empty() {
            return None;
        }
        let index = NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed) % registry.len();
        registry
            .values()
            .nth(index)
            .filter(|(_, seen)| seen.elapsed() < Duration::from_secs(86400))
            .map(|(app, _)| app.clone())
    });
    if let Some(preview) = preview {
        #[cfg(not(target_os = "wasi"))]
        super::social_connections::process_one(&preview);
        process_one_source(&preview);
    }
}

/// Verified event facts or an attributed public day-only announcement; unrelated feed prose is excluded.
#[derive(Debug, Clone, PartialEq)]
struct DiscoveredEvent {
    key: String,
    title: String,
    starts_at: String,
    ends_at: String,
    venue: String,
    address: String,
    country: String,
    timezone: String,
    event_state: String,
    state_explicit: bool,
    organizers: Vec<String>,
    event_url: String,
    ticket_url: String,
    status: String,
    evidence: String,
    evidence_source: String,
}

/// Retains inert source facts with a host-authored reason for private owner review.
#[derive(Clone)]
struct ReviewCandidate {
    facts: Value,
    reason: &'static str,
    accepted_event_id: Option<i64>,
}

/// Accepts bounded HTTPS links without credentials, controls or executable schemes.
fn public_link(value: &str) -> Option<String> {
    Some(super::outbound_url::https_url(value)?.to_string())
}

/// Loads bounded active organizer sites; storage/capacity failures never retire sources or widen authority.
fn organizer_sites(store: &super::database::LogicalAppStore) -> Result<Vec<url::Url>, ()> {
    if !store
        .schema
        .entities
        .iter()
        .any(|entity| entity.name == "organizer")
    {
        return Ok(Vec::new());
    }
    let rows = store
        .adapter
        .query(
            &store.config,
            &store.schema,
            &Query {
                entity: "organizer".into(),
                fields: vec!["website".into()],
                filter: Some(Predicate::Equal("crm_status".into(), json!("active"))),
                order: vec![],
                limit: 1000,
                offset: 0,
            },
        )
        .map_err(|_| ())?;
    if rows.len() == 1000 {
        return Err(());
    }
    Ok(rows
        .iter()
        .filter_map(|row| {
            row.get("website")
                .and_then(Value::as_str)
                .and_then(super::outbound_url::https_url)
        })
        .collect())
}

/// Compares declared site origins while tolerating only the conventional www alias.
fn same_site(left: &url::Url, right: &url::Url) -> bool {
    left.host_str()
        .map(|host| host.strip_prefix("www.").unwrap_or(host))
        == right
            .host_str()
            .map(|host| host.strip_prefix("www.").unwrap_or(host))
        && left.port_or_known_default() == right.port_or_known_default()
}

/// Treats social hosts as profile scopes even when pasted into a website field.
fn declared_site_allows(target: &url::Url, site: &url::Url) -> bool {
    if !same_site(target, site) {
        return false;
    }
    let host = site.host_str().unwrap_or_default();
    let social = requires_social_authorization(site)
        || ["youtube.com", "youtu.be", "soundcloud.com", "spotify.com"]
            .iter()
            .any(|domain| {
                host == *domain
                    || host
                        .strip_suffix(domain)
                        .is_some_and(|prefix| prefix.ends_with('.'))
            });
    if !social {
        return true;
    }
    let path = site.path().trim_end_matches('/');
    !path.is_empty()
        && target.query() == site.query()
        && (target.path().trim_end_matches('/') == path
            || target.path().starts_with(&format!("{path}/")))
}

/// Limits discovery to explicitly declared artist/social/organizer sites, excluding catalog provenance and link hubs.
fn eligible_source(value: &str, artist: &Map<String, Value>, organizers: &[url::Url]) -> bool {
    let Some(url) = super::outbound_url::https_url(value) else {
        return false;
    };
    let host = url.host_str().unwrap_or_default();
    if [
        "bing.com",
        "google.com",
        "duckduckgo.com",
        "search.yahoo.com",
        "linktr.ee",
        "linktree.com",
    ]
    .iter()
    .any(|blocked| {
        host == *blocked
            || host
                .strip_suffix(blocked)
                .is_some_and(|prefix| prefix.ends_with('.'))
    }) {
        return false;
    }
    if artist
        .get("source_url")
        .and_then(Value::as_str)
        .and_then(super::outbound_url::https_url)
        .is_some_and(|catalog| {
            catalog
                .path()
                .split('/')
                .any(|part| matches!(part, "artists" | "artist" | "djs" | "dj"))
                && same_site(&url, &catalog)
                && (url.path().trim_end_matches('/') == catalog.path().trim_end_matches('/')
                    || url
                        .path()
                        .starts_with(&format!("{}/", catalog.path().trim_end_matches('/'))))
        })
    {
        return false;
    }
    if artist
        .get("website")
        .and_then(Value::as_str)
        .and_then(super::outbound_url::https_url)
        .is_some_and(|site| declared_site_allows(&url, &site))
        || organizers
            .iter()
            .any(|site| declared_site_allows(&url, site))
    {
        return true;
    }
    PROFILE_FIELDS
        .iter()
        .filter(|field| !matches!(**field, "website" | "source_url"))
        .any(|field| {
            artist
                .get(*field)
                .and_then(Value::as_str)
                .and_then(super::outbound_url::https_url)
                .is_some_and(|profile| {
                    let path = profile.path().trim_end_matches('/');
                    !path.is_empty()
                        && same_site(&url, &profile)
                        && url.query() == profile.query()
                        && (url.path().trim_end_matches('/') == path
                            || url.path().starts_with(&format!("{path}/")))
                })
        })
}

/// Matches performer names without guessing identities from partial substrings.
fn normalized_name(value: &str) -> String {
    value
        .split_whitespace()
        .collect::<Vec<_>>()
        .join(" ")
        .to_lowercase()
}

/// Collects performer names from the bounded JSON-LD value.
fn performer_matches(value: &Value, name: &str) -> bool {
    match value {
        Value::String(value) => normalized_name(value) == normalized_name(name),
        Value::Array(values) => values.iter().any(|value| performer_matches(value, name)),
        Value::Object(value) => value
            .get("name")
            .is_some_and(|value| performer_matches(value, name)),
        _ => false,
    }
}

/// Extracts one complete event, requiring an explicit time zone, performer and venue.
fn event_from_value(
    value: &Value,
    artist: &str,
    source: &str,
    now: u64,
) -> Option<DiscoveredEvent> {
    let types = value.get("@type")?;
    let is_event = |kind: &str| {
        matches!(
            kind.rsplit('/').next().unwrap_or(kind),
            "Event" | "MusicEvent" | "Festival"
        )
    };
    if !types.as_str().is_some_and(is_event)
        && !types
            .as_array()
            .is_some_and(|types| types.iter().filter_map(Value::as_str).any(is_event))
    {
        return None;
    }
    if !performer_matches(value.get("performer")?, artist) {
        return None;
    }
    let title = value.get("name")?.as_str()?.trim();
    let raw_start = value.get("startDate")?.as_str()?;
    if raw_start.len() < 20
        || !(raw_start.ends_with('Z') || raw_start.get(19..)?.contains(['+', '-']))
    {
        return None;
    }
    let start = runtime_timestamp_seconds(raw_start)?;
    if start < now.saturating_sub(30 * 86400)
        || start > now.saturating_add(730 * 86400)
        || title.is_empty()
        || title.len() > 300
    {
        return None;
    }
    let venue = value.pointer("/location/name")?.as_str()?.trim();
    if venue.is_empty() || venue.len() > 300 {
        return None;
    }
    let event_url = value
        .get("url")
        .and_then(Value::as_str)
        .and_then(public_link)
        .unwrap_or_else(|| source.to_owned());
    let ticket = value
        .pointer("/offers/url")
        .or_else(|| value.pointer("/offers/0/url"))
        .and_then(Value::as_str)
        .and_then(public_link)
        .unwrap_or_default();
    let end = if let Some(raw) = value.get("endDate") {
        let raw = raw.as_str()?;
        if raw.len() < 20 || !(raw.ends_with('Z') || raw.get(19..)?.contains(['+', '-'])) {
            return None;
        }
        let end = runtime_timestamp_seconds(raw)?;
        if end < start {
            return None;
        }
        pharos_runtime_core::time::timestamp_from_epoch_seconds(end)
    } else {
        String::new()
    };
    let (status, event_state) = match value
        .get("eventStatus")
        .and_then(Value::as_str)
        .unwrap_or("")
        .rsplit('/')
        .next()?
    {
        "EventCancelled" => ("cancelled", "cancelled"),
        "EventPostponed" => ("review", "postponed"),
        "EventRescheduled" => ("published", "rescheduled"),
        "EventScheduled" | "" => ("published", "scheduled"),
        _ => ("review", "unknown"),
    };
    let key = model::event_identity(value, source, start, venue, title, &event_url)?;
    let evidence = json!({"extraction_version":1,"event":value}).to_string();
    if evidence.len() > 7680 {
        return None;
    }
    let (address, country) = model::location(value)?;
    let organizers = model::organizer_names(value)?;
    Some(DiscoveredEvent {
        key,
        title: title.to_owned(),
        starts_at: pharos_runtime_core::time::timestamp_from_epoch_seconds(start),
        ends_at: end,
        venue: venue.to_owned(),
        address,
        country,
        timezone: if raw_start.ends_with('Z') {
            "UTC".into()
        } else {
            raw_start.get(raw_start.len().checked_sub(6)?..)?.to_owned()
        },
        event_state: event_state.into(),
        state_explicit: value
            .get("eventStatus")
            .and_then(Value::as_str)
            .is_some_and(|state| !state.is_empty()),
        organizers,
        event_url,
        ticket_url: ticket,
        status: status.to_owned(),
        evidence,
        evidence_source: source.to_owned(),
    })
}

/// Traverses only bounded JSON-LD graphs; arbitrary nested documents are not crawled.
fn collect_events(
    value: &Value,
    artist: &str,
    source: &str,
    now: u64,
    depth: usize,
    output: &mut BTreeMap<String, DiscoveredEvent>,
    review: &mut Vec<ReviewCandidate>,
) {
    if depth > 8 || output.len() + review.len() >= 100 {
        return;
    }
    let event = event_from_value(value, artist, source, now);
    let in_window = value
        .get("startDate")
        .and_then(Value::as_str)
        .and_then(runtime_timestamp_seconds)
        .is_none_or(|start| {
            start >= now.saturating_sub(30 * 86400) && start <= now.saturating_add(730 * 86400)
        });
    if event.is_none()
        && in_window
        && value.get("@type").is_some_and(|kind| {
            let event_type = |name: &str| {
                matches!(
                    name.rsplit('/').next(),
                    Some("Event" | "MusicEvent" | "Festival")
                )
            };
            kind.as_str().is_some_and(event_type)
                || kind
                    .as_array()
                    .is_some_and(|kinds| kinds.iter().filter_map(Value::as_str).any(event_type))
        })
        && value.to_string().len() <= 7680
    {
        review.push(ReviewCandidate {
            facts: value.clone(),
            reason: "Incomplete date, venue or exact artist evidence; not published",
            accepted_event_id: None,
        });
    }
    if let Some(event) = event {
        match output.entry(event.key.clone()) {
            std::collections::btree_map::Entry::Occupied(mut entry) => {
                if entry.get() != &event {
                    entry.get_mut().status = "review".to_owned();
                    review.push(ReviewCandidate {
                        facts: value.clone(),
                        reason: "Conflicting event declarations; not published",
                        accepted_event_id: None,
                    });
                }
            }
            std::collections::btree_map::Entry::Vacant(entry) => {
                entry.insert(event);
            }
        }
    }
    match value {
        Value::Array(values) => {
            for value in values.iter().take(100) {
                collect_events(value, artist, source, now, depth + 1, output, review);
            }
        }
        Value::Object(values) => {
            if let Some(graph) = values.get("@graph") {
                collect_events(graph, artist, source, now, depth + 1, output, review);
            }
        }
        _ => {}
    }
}

/// Reads application/ld+json blocks or a JSON event feed without executing source markup.
#[cfg(test)]
fn extract_events(body: &str, artist: &str, source: &str, now: u64) -> Vec<DiscoveredEvent> {
    extract_events_with_review(body, artist, source, now).0
}

/// Retains bounded incomplete structured evidence privately; only validated facts become public events.
fn extract_events_with_review(
    body: &str,
    artist: &str,
    source: &str,
    now: u64,
) -> (Vec<DiscoveredEvent>, Vec<ReviewCandidate>) {
    if body.len() > 1024 * 1024 || public_link(source).is_none() {
        return (Vec::new(), Vec::new());
    }
    let mut output = BTreeMap::new();
    if body.trim_start().starts_with("BEGIN:VCALENDAR") {
        return (
            calendar_feed::extract(body, artist, source, now).unwrap_or_default(),
            Vec::new(),
        );
    }
    let mut review = Vec::new();
    if let Ok(value) = serde_json::from_str::<Value>(body) {
        collect_events(&value, artist, source, now, 0, &mut output, &mut review);
    } else {
        let lowercase = body.to_ascii_lowercase();
        let mut lower_remainder = lowercase.as_str();
        let mut remainder = body;
        for _ in 0..100 {
            let Some(start) = lower_remainder.find("<script") else {
                break;
            };
            remainder = &remainder[start + 7..];
            lower_remainder = &lower_remainder[start + 7..];
            let Some(end_tag) = lower_remainder.find('>') else {
                break;
            };
            let header = &remainder[..end_tag];
            remainder = &remainder[end_tag + 1..];
            lower_remainder = &lower_remainder[end_tag + 1..];
            let Some(end) = lower_remainder.find("</script>") else {
                break;
            };
            if header.to_ascii_lowercase().contains("application/ld+json")
                && let Ok(value) = serde_json::from_str::<Value>(&remainder[..end])
            {
                collect_events(&value, artist, source, now, 0, &mut output, &mut review);
            }
            remainder = &remainder[end + 9..];
            lower_remainder = &lower_remainder[end + 9..];
        }
    }
    (output.into_values().collect(), review)
}

/// Builds literal values for parameterized generic database mutations.
fn literals(values: Value) -> BTreeMap<String, WriteValue> {
    values
        .as_object()
        .into_iter()
        .flatten()
        .map(|(key, value)| (key.clone(), WriteValue::Literal(value.clone())))
        .collect()
}

/// Removes stale or unattributed social review material without changing accepted gigs or owner decisions.
fn prune_social_reviews(store: &super::database::LogicalAppStore, now: u64) -> Result<(), ()> {
    if !store
        .schema
        .entities
        .iter()
        .any(|entity| entity.name == "event_observation")
    {
        return Ok(());
    }
    let rows = store
        .adapter
        .query(
            &store.config,
            &store.schema,
            &Query {
                entity: "event_observation".into(),
                fields: vec!["id".into(), "source_id".into(), "facts".into()],
                filter: Some(Predicate::Equal("review_status".into(), json!("review"))),
                order: vec![SortKey {
                    field: "id".into(),
                    descending: false,
                }],
                limit: 1000,
                offset: 0,
            },
        )
        .map_err(|_| ())?;
    let mut steps = Vec::new();
    for row in rows {
        let raw = row.get("facts").and_then(Value::as_str).ok_or(())?;
        let facts: Value = serde_json::from_str(raw).map_err(|_| ())?;
        if facts.get("accepted_event_id").is_some() {
            continue;
        }
        let Some(candidate) = facts.get("candidate") else {
            continue;
        };
        let social =
            candidate.get("evidence_kind").is_some() || candidate.get("source_profile").is_some();
        if !social || social_evidence::eligible(candidate, now) {
            continue;
        }
        // Retain deduplication identity, not leaked account panels or unrelated personal-stream prose.
        steps.push(Mutation::Update {entity:"event_observation".into(),filter:Predicate::All(vec![Predicate::Equal("id".into(),row["id"].clone()),Predicate::Equal("source_id".into(),row["source_id"].clone()),Predicate::Equal("facts".into(),row["facts"].clone()),Predicate::Equal("review_status".into(),json!("review"))]),values:literals(json!({"review_status":"rejected","facts":json!({"extraction_version":2,"review_reason":"Excluded: stale, retrospective or unverified post attribution/date evidence","candidate":{}}).to_string()})),expected_rows:Some(1)});
    }
    for batch in steps.chunks(100) {
        store
            .adapter
            .transact(
                &store.config,
                &store.schema,
                &Transaction {
                    steps: batch.to_vec(),
                    idempotency_key: None,
                },
            )
            .map_err(|_| ())?;
    }
    Ok(())
}

/// Deduplicates private evidence using bounded batches matching the shared predicate budget.
fn pending_review_evidence(
    store: &super::database::LogicalAppStore,
    source: i64,
    candidates: Vec<ReviewCandidate>,
) -> Result<Vec<(String, String)>, ()> {
    let mut evidence = BTreeMap::new();
    for candidate in candidates {
        let mut facts = json!({"extraction_version":1,"candidate":candidate.facts,"review_reason":candidate.reason});
        if let Some(id) = candidate.accepted_event_id {
            facts["accepted_event_id"] = json!(id);
        }
        let facts = facts.to_string();
        if facts.len() > 8192 {
            return Err(());
        }
        let hash =
            pharos_runtime_database::operations::operation_fingerprint(&facts).map_err(|_| ())?;
        evidence.insert(hash, facts);
    }
    if evidence.is_empty() {
        return Ok(Vec::new());
    }
    let hashes = evidence.keys().cloned().collect::<Vec<_>>();
    let mut rows = Vec::new();
    for batch in hashes.chunks(64) {
        rows.extend(
            store
                .adapter
                .query(
                    &store.config,
                    &store.schema,
                    &Query {
                        entity: "event_observation".into(),
                        fields: vec!["evidence_hash".into(), "review_status".into()],
                        filter: Some(Predicate::All(vec![
                            Predicate::Equal("source_id".into(), json!(source)),
                            Predicate::In(
                                "evidence_hash".into(),
                                batch.iter().map(|hash| json!(hash)).collect(),
                            ),
                        ])),
                        order: vec![],
                        limit: 64,
                        offset: 0,
                    },
                )
                .map_err(|_| ())?,
        );
    }
    for row in rows {
        if row
            .get("review_status")
            .and_then(Value::as_str)
            .is_some_and(|status| matches!(status, "rejected" | "reviewed" | "accepted"))
            && let Some(hash) = row.get("evidence_hash").and_then(Value::as_str)
        {
            evidence.remove(hash);
        }
    }
    Ok(evidence
        .into_iter()
        .map(|(hash, facts)| (facts, hash))
        .collect())
}

/// Reads one declared source entity without backend-specific persistence.
fn query(
    store: &super::database::LogicalAppStore,
    entity: &str,
    fields: &[&str],
    filter: Predicate,
) -> Option<Map<String, Value>> {
    store
        .adapter
        .query(
            &store.config,
            &store.schema,
            &Query {
                entity: entity.to_owned(),
                fields: fields.iter().map(|field| (*field).to_owned()).collect(),
                filter: Some(filter),
                order: vec![],
                limit: 1,
                offset: 0,
            },
        )
        .ok()?
        .into_iter()
        .next()
}

/// Synchronizes one artist's profile links using a durable scan cursor and bounded writes.
/// Existing sources retain an owner's pause state; catalog row identities are resolved from storage.
fn synchronize_one_artist(
    store: &super::database::LogicalAppStore,
    organizers: &[url::Url],
) -> bool {
    let scope = Predicate::Equal("scope_key".into(), json!("artist_source_scan"));
    let counter = query(store, "crm_counter", &["last_value"], scope.clone());
    let cursor = counter
        .as_ref()
        .and_then(|row| row.get("last_value"))
        .and_then(Value::as_i64)
        .unwrap_or(0);
    let Ok(rows) = store.adapter.query(
        &store.config,
        &store.schema,
        &Query {
            entity: "artists".into(),
            fields: std::iter::once("csv_row_id")
                .chain(PROFILE_FIELDS)
                .map(str::to_owned)
                .collect(),
            filter: Some(Predicate::All(vec![
                Predicate::GreaterThan("csv_row_id".into(), json!(cursor)),
                Predicate::Equal("crm_status".into(), json!("active")),
            ])),
            order: vec![SortKey {
                field: "csv_row_id".into(),
                descending: false,
            }],
            limit: 1,
            offset: 0,
        },
    ) else {
        return false;
    };
    let next = rows
        .first()
        .and_then(|row| row.get("csv_row_id"))
        .and_then(Value::as_i64)
        .unwrap_or(0);
    if cursor == 0 && next == 0 {
        return false;
    }
    let guard = if counter.is_some() {
        Predicate::All(vec![
            scope.clone(),
            Predicate::Equal("last_value".into(), json!(cursor)),
        ])
    } else {
        scope
    };
    let mut steps = vec![
        Mutation::RequireCount {
            entity: "crm_counter".into(),
            filter: guard,
            minimum: u64::from(counter.is_some()),
            maximum: Some(u64::from(counter.is_some())),
        },
        Mutation::Upsert {
            entity: "crm_counter".into(),
            unique_key: vec!["scope_key".into()],
            values: literals(json!({"scope_key":"artist_source_scan","last_value":next})),
        },
    ];
    if let Some(artist) = rows.first() {
        let stale = match store.adapter.query(
            &store.config,
            &store.schema,
            &Query {
                entity: "event_source".into(),
                fields: vec!["id".into(), "url".into(), "last_status".into()],
                filter: Some(Predicate::All(vec![
                    Predicate::Equal("artist_row_id".into(), json!(next)),
                    Predicate::NotEqual("last_status".into(), json!("source_removed")),
                ])),
                order: vec![],
                limit: 1000,
                offset: 0,
            },
        ) {
            Ok(rows) => rows,
            Err(_) => return false,
        };
        let mut artist_guard = vec![
            Predicate::Equal("csv_row_id".into(), json!(next)),
            Predicate::Equal("crm_status".into(), json!("active")),
        ];
        for field in PROFILE_FIELDS {
            artist_guard.push(Predicate::Equal(field.into(), artist[field].clone()));
        }
        steps.push(Mutation::RequireCount {
            entity: "artists".into(),
            filter: Predicate::All(artist_guard),
            minimum: 1,
            maximum: Some(1),
        });
        for source in stale.iter().filter(|source| {
            !source
                .get("url")
                .and_then(Value::as_str)
                .is_some_and(|value| eligible_source(value, artist, organizers))
        }) {
            steps.push(Mutation::Update {entity:"event_source".into(),filter:Predicate::All(vec![Predicate::Equal("id".into(),source["id"].clone()),Predicate::Equal("url".into(),source["url"].clone()),Predicate::Equal("last_status".into(),source["last_status"].clone())]),values:literals(json!({"enabled":false,"last_status":"source_removed","last_message":"Not a configured artist website, social profile or active organizer site"})),expected_rows:Some(1)});
        }
        for name in PROFILE_FIELDS
            .into_iter()
            .filter(|name| *name != "source_url")
        {
            let Some(url) = artist
                .get(name)
                .and_then(Value::as_str)
                .and_then(public_link)
                .filter(|url| url.len() <= 512 && eligible_source(url, artist, organizers))
            else {
                continue;
            };
            steps.push(Mutation::InsertIfAbsent {entity:"event_source".into(),unique_key:vec!["artist_row_id".into(),"url".into()],values:literals(json!({"artist_row_id":next,"url":url,"enabled":true,"managed":true,"next_check_at":current_timestamp_utc(),"last_status":"pending","last_checked_at":"","last_message":""}))});
        }
    }
    if let Err(error) = store.adapter.transact(
        &store.config,
        &store.schema,
        &Transaction {
            steps,
            idempotency_key: None,
        },
    ) {
        eprintln!("event source synchronization failed: {error:?}");
        return false;
    }
    next != 0
}

/// Fetches one public HTTPS document with bounded response bytes and no redirects.
fn fetch(url: &str) -> Option<pharos_runtime_integrations::webhook::WebhookResponse> {
    let mut target = super::outbound_url::https_url(url)?;
    target.set_fragment(None);
    pharos_runtime_integrations::webhook::send_request(
        "GET",
        target.as_str(),
        None,
        "",
        &[
            ("User-Agent", "PharosEventDiscovery/0.8.4".to_owned()),
            (
                "Accept",
                "application/ld+json, application/json, text/calendar, text/html".to_owned(),
            ),
        ],
        pharos_runtime_integrations::webhook::WebhookLimits {
            connect_timeout: Duration::from_secs(3),
            operation_timeout: Duration::from_secs(8),
            max_response_bytes: 1024 * 1024,
            allow_plaintext_local: false,
        },
    )
}

/// Applies conservative public robots rules before reading an explicitly configured page.
/// Redirects or unavailable policies fail closed; a missing robots document permits access.
fn robots_allowed(url: &str) -> bool {
    let Some(parsed) = super::outbound_url::https_url(url) else {
        return false;
    };
    let Ok(robots) = parsed.join("/robots.txt") else {
        return false;
    };
    let Some(response) = fetch(robots.as_str()) else {
        return false;
    };
    if response.status_code == 404 {
        return true;
    }
    if response.status_code != 200 || response.body.len() > 65536 {
        return false;
    }
    let path = parsed.query().map_or_else(
        || parsed.path().to_owned(),
        |query| format!("{}?{query}", parsed.path()),
    );
    robots_body_allows(&path, &response.body)
}

/// Evaluates bounded literal robots rules; unsupported encoded or wildcard policies fail closed.
/// This intentionally supports a narrower public-source policy than a general web crawler.
fn robots_body_allows(path: &str, body: &str) -> bool {
    if path.contains('%') || body.len() > 65536 || body.lines().count() > 2000 {
        return false;
    }
    let mut applies = false;
    let mut in_rules = false;
    let mut allow = 0;
    let mut deny = 0;
    for line in body.lines() {
        let line = line.split('#').next().unwrap_or_default().trim();
        let Some((field, value)) = line.split_once(':') else {
            continue;
        };
        let value = value.trim();
        match field.trim().to_ascii_lowercase().as_str() {
            "user-agent" => {
                if in_rules {
                    applies = false;
                    in_rules = false;
                }
                applies |= value == "*"
                    || value
                        .to_ascii_lowercase()
                        .starts_with("pharoseventdiscovery");
            }
            kind @ ("allow" | "disallow") => {
                in_rules = true;
                if !applies || value.is_empty() {
                    continue;
                }
                if value.contains(['*', '$', '%']) {
                    return false;
                }
                let prefix = value;
                if path.starts_with(prefix) {
                    if kind == "allow" {
                        allow = allow.max(prefix.len());
                    } else {
                        deny = deny.max(prefix.len());
                    }
                }
            }
            "crawl-delay" | "request-rate" if applies => return false,
            _ => {}
        }
    }
    deny == 0 || allow >= deny
}

/// Distinguishes a source policy refusal from a bounded transport failure in owner feedback.
#[derive(Debug, Clone, Copy)]
enum SourceReadError {
    AccessBlocked,
    Unavailable,
}

/// Claims at most one due source per poll and records actual fetch/acceptance outcomes.
pub(super) fn process_one_source(app: &AppRuntime) {
    if !app.queue_enabled
        || dynamic::dynamic_contract_doc(app, "event_discovery_contract")
            .and_then(|value| value.get("enabled").and_then(Value::as_bool))
            != Some(true)
    {
        return;
    }
    let Some(store) = app_database_operations(app) else {
        return;
    };
    process_store_source_with_app(Some(app), &store, |url| {
        if robots_allowed(url) {
            fetch(url).ok_or(SourceReadError::Unavailable)
        } else {
            Err(SourceReadError::AccessBlocked)
        }
    });
}

/// Claims and accepts one source using the generic store; tests can supply recorded public responses.
#[cfg(test)]
fn process_store_source(
    store: &super::database::LogicalAppStore,
    read: impl FnOnce(
        &str,
    )
        -> Result<pharos_runtime_integrations::webhook::WebhookResponse, SourceReadError>,
) {
    process_store_source_with_app(None, store, read);
}

/// Shares durable source acceptance between public-page tests and authorized native consumers.
fn process_store_source_with_app(
    app: Option<&AppRuntime>,
    store: &super::database::LogicalAppStore,
    read: impl FnOnce(
        &str,
    )
        -> Result<pharos_runtime_integrations::webhook::WebhookResponse, SourceReadError>,
) {
    let Ok(approved_organizers) = organizer_sites(store) else {
        return;
    };
    for _ in 0..16 {
        if !synchronize_one_artist(store, &approved_organizers) {
            break;
        }
    }
    let now = current_timestamp_utc();
    if prune_social_reviews(store, runtime_timestamp_seconds(&now).unwrap_or(u64::MAX)).is_err() {
        return;
    }
    let due = Predicate::All(vec![
        Predicate::Equal("enabled".to_owned(), json!(true)),
        Predicate::AtMost("next_check_at".to_owned(), json!(now)),
    ]);
    let Ok(rows) = store.adapter.query(
        &store.config,
        &store.schema,
        &Query {
            entity: "event_source".to_owned(),
            fields: vec![
                "id".to_owned(),
                "artist_row_id".to_owned(),
                "url".to_owned(),
                "next_check_at".to_owned(),
                "managed".to_owned(),
            ],
            filter: Some(due),
            order: vec![SortKey {
                field: "next_check_at".to_owned(),
                descending: false,
            }],
            limit: 1,
            offset: 0,
        },
    ) else {
        return;
    };
    let Some(source) = rows.first() else {
        return;
    };
    let (Some(id), Some(artist_id), Some(url)) = (
        source.get("id").and_then(Value::as_i64),
        source.get("artist_row_id").and_then(Value::as_i64),
        source.get("url").and_then(Value::as_str),
    ) else {
        return;
    };
    let source_filter = Predicate::All(vec![
        Predicate::Equal("id".into(), json!(id)),
        Predicate::Equal("next_check_at".into(), source["next_check_at"].clone()),
        Predicate::Equal("enabled".into(), json!(true)),
    ]);
    let lease = pharos_runtime_core::time::timestamp_after_seconds_utc(120);
    if store.adapter.transact(&store.config, &store.schema, &Transaction {steps:vec![Mutation::Update {entity:"event_source".into(),filter:source_filter,values:literals(json!({"next_check_at":lease,"last_status":"running","last_checked_at":now})),expected_rows:Some(1)}],idempotency_key:None}).is_err() { return; }
    let artist_record = query(
        store,
        "artists",
        &std::iter::once("name")
            .chain(["crm_status", "record_updated_at"])
            .chain(PROFILE_FIELDS)
            .collect::<Vec<_>>(),
        Predicate::Equal("csv_row_id".into(), json!(artist_id)),
    );
    let artist = artist_record
        .as_ref()
        .filter(|row| row.get("crm_status").and_then(Value::as_str) == Some("active"))
        .and_then(|row| row.get("name"))
        .and_then(Value::as_str)
        .filter(|name| !name.trim().is_empty())
        .map(str::to_owned);
    let removed = artist.is_none()
        || !artist_record
            .as_ref()
            .is_some_and(|record| eligible_source(url, record, &approved_organizers));
    let parsed_url = super::outbound_url::https_url(url);
    let social_authorization = parsed_url
        .as_ref()
        .is_some_and(requires_social_authorization);
    let allowed = !removed && artist.is_some() && !social_authorization && parsed_url.is_some();
    let response = if allowed {
        read(url)
    } else {
        Err(SourceReadError::AccessBlocked)
    };
    let social_result = if social_authorization && !removed {
        app.map(|app| {
            queue_social_links(
                app,
                store,
                url,
                artist_id,
                Predicate::All(vec![
                    Predicate::Equal("id".into(), json!(id)),
                    Predicate::Equal("next_check_at".into(), json!(lease)),
                    Predicate::Equal("url".into(), json!(url)),
                    Predicate::Equal("enabled".into(), json!(true)),
                ]),
                artist_record
                    .as_ref()
                    .and_then(|row| row.get("record_updated_at"))
                    .cloned()
                    .unwrap_or(Value::Null),
            )
        })
    } else {
        None
    };
    let supports_review = store.schema.entities.iter().any(|entity| {
        entity.name == "event_observation"
            && entity
                .fields
                .iter()
                .any(|field| field.name == "review_status")
    });
    let mut reviews = Vec::new();
    let (mut status, mut detail, events) = match (response, artist) {
        (Ok(response), _) if response.body.len() > 1024 * 1024 => (
            "failed", "Source response exceeded the intake limit".to_owned(), vec![],
        ),
        (Ok(response), Some(artist)) if response.status_code == 200 => {
            let epoch = runtime_timestamp_seconds(&now).unwrap_or(u64::MAX);
            let extracted = if response.body.trim_start().starts_with("BEGIN:VCALENDAR") {
                calendar_feed::extract(&response.body, &artist, url, epoch).map(|events|(events,Vec::new()))
            } else {
                Ok(extract_events_with_review(&response.body, &artist, url, epoch))
            };
            match extracted {
                Ok((events,unaccepted)) => {
                    reviews=unaccepted;
                    if events.is_empty() && reviews.is_empty() && password_gate(&response.body) {
                        ("authorization_required","Source returned a login page rather than an event feed".into(),Vec::new())
                    } else {
                    (
                    if !reviews.is_empty() {"review"} else if events.is_empty() { "no_events" } else { "success" },
                    format!("{} verified structured events; {} candidates need owner review", events.len(),reviews.len()),
                    events,
                )}},
                Err(detail) => ("unsupported_source", detail.to_owned(), vec![]),
            }
        }
        (Ok(response), _) if matches!(response.status_code,404|410) => (
            "not_found", "Source reports that the account or page does not exist or has been removed".to_owned(), vec![],
        ),
        (Ok(response), _) if matches!(response.status_code, 401 | 403) => (
            "authorization_required",
            "Source requires provider authorization".to_owned(),
            vec![],
        ),
        (Ok(response), _) if response.status_code == 429 => (
            "rate_limited",
            "Source rate limited this request".to_owned(),
            vec![],
        ),
        (Ok(response), _) => (
            "failed",
            format!("Source returned HTTP {}", response.status_code),
            vec![],
        ),
        _ if removed => (
            "source_removed",
            "Source is not a configured artist website, social profile or active organizer site".to_owned(),
            vec![],
        ),
        _ if social_result.as_ref().is_some_and(|result|result.as_ref().is_ok_and(|intake|intake.unverified_posts>0 && intake.verified_posts==0)) => (
            "unsupported_source",
            "Posts were visible, but their artist attribution or actual posting dates could not be verified. Unverified posts were excluded; no events were inferred from feed text".to_owned(),
            vec![],
        ),
        _ if social_result.as_ref().is_some_and(Result::is_ok) => {
            let intake = social_result.as_ref().and_then(|result| result.as_ref().ok());
            let events = intake.map(|intake| intake.events.clone()).unwrap_or_default();
            (if events.is_empty() {"social_collected"} else {"success"},
            format!("{} verified artist performances; {} public source links retained; {} posts have verified posting dates; partial browser view, not a complete feed",events.len(),intake.map_or(0,|intake|intake.link_count),intake.map_or(0,|intake|intake.verified_posts)), events)
        },
        _ if social_result.as_ref().is_some_and(|result|matches!(result,Err(pharos_runtime_social::SocialError::Busy))) => (
            "pending",
            "Authorized social collection is queued".into(),
            vec![],
        ),
        _ if social_result.as_ref().is_some_and(|result|matches!(result,Err(pharos_runtime_social::SocialError::RateLimited))) => (
            "rate_limited",
            "Provider rate limit reached; collection will retry".into(),
            vec![],
        ),
        _ if social_result.as_ref().is_some_and(|result|matches!(result,Err(error) if !matches!(error,pharos_runtime_social::SocialError::AuthorizationRequired|pharos_runtime_social::SocialError::UnsupportedProvider|pharos_runtime_social::SocialError::Forbidden))) => (
            "failed",
            social_result.as_ref().and_then(|result|result.as_ref().err()).map(ToString::to_string).unwrap_or_default(),
            vec![],
        ),
        _ if social_result.as_ref().is_some_and(|result|matches!(result,Err(pharos_runtime_social::SocialError::UnsupportedProvider))) => (
            "connector_unavailable",
            pharos_runtime_social::SocialError::UnsupportedProvider.to_string(),
            vec![],
        ),
        _ if social_authorization => (
            "authorization_required",
            "This social source needs an authorized provider connection".to_owned(),
            vec![],
        ),
        (Err(SourceReadError::AccessBlocked), _) => (
            "access_blocked",
            "Public crawler access was not permitted or its robots policy could not be verified. The page was not fetched; this does not establish whether the account or page exists".to_owned(),
            vec![],
        ),
        _ => (
            "failed",
            "Source could not be read safely within the fetch limits".to_owned(),
            vec![],
        ),
    };
    if let Some(Ok(intake)) = &social_result {
        reviews.extend(intake.reviews.clone());
    }
    let Ok(existing_events) = model::existing_events(store, &events) else {
        return;
    };
    let mut accepted = Vec::new();
    for event in events {
        if let Some(existing) = existing_events.get(&event.key)
            && model::conflicts(existing, &event, id)
        {
            let Ok(facts) = serde_json::from_str(&event.evidence) else {
                return;
            };
            reviews.push(ReviewCandidate {facts,reason:"Conflicts with an accepted event; compare retained evidence before updating it",accepted_event_id:existing.get("id").and_then(Value::as_i64)});
        } else {
            accepted.push(event);
        }
    }
    let events = accepted;
    if !reviews.is_empty() {
        status = "review";
    }
    let review_required = !reviews.is_empty();
    let reviews = if supports_review {
        let Ok(pending) = pending_review_evidence(store, id, reviews) else {
            return;
        };
        pending
    } else {
        Vec::new()
    };
    if status == "review" {
        status = if review_required && !supports_review {
            "unsupported_source"
        } else if reviews.is_empty() {
            if events.is_empty() {
                if review_required {
                    "reviewed"
                } else {
                    "no_events"
                }
            } else {
                "success"
            }
        } else {
            "review"
        };
        detail = if review_required && !supports_review {
            "Source needs owner review; this app has no review contract".into()
        } else {
            format!(
                "{} verified structured events; {} candidates need owner review",
                events.len(),
                reviews.len()
            )
        };
    }
    let Ok(organizers) = (if events.iter().any(|event| !event.organizers.is_empty()) {
        model::known_organizers(store)
    } else {
        Ok(BTreeMap::new())
    }) else {
        return;
    };
    let mut common_guards = vec![Mutation::RequireCount {
        entity: "event_source".into(),
        filter: Predicate::All(vec![
            Predicate::Equal("id".into(), json!(id)),
            Predicate::Equal("next_check_at".into(), json!(lease)),
            Predicate::Equal("url".into(), json!(url)),
            Predicate::Equal("enabled".into(), json!(true)),
        ]),
        minimum: 1,
        maximum: Some(1),
    }];
    if let Some(Ok(intake)) = &social_result {
        common_guards.extend(intake.guards.clone());
    }
    if !events.is_empty() || !reviews.is_empty() {
        let Some(artist) = &artist_record else {
            return;
        };
        common_guards.push(Mutation::RequireCount {
            entity: "artists".into(),
            filter: Predicate::All(vec![
                Predicate::Equal("csv_row_id".into(), json!(artist_id)),
                Predicate::Equal("crm_status".into(), json!("active")),
                Predicate::Equal(
                    "record_updated_at".into(),
                    artist["record_updated_at"].clone(),
                ),
            ]),
            minimum: 1,
            maximum: Some(1),
        });
    }
    let mut groups = Vec::new();
    if supports_review {
        for (facts, evidence_hash) in reviews {
            let Ok(event_key) =
                pharos_runtime_database::operations::operation_fingerprint(&(url, &evidence_hash))
            else {
                continue;
            };
            groups.push(vec![Mutation::InsertIfAbsent {entity:"event_observation".into(),unique_key:vec!["source_id".into(),"evidence_hash".into()],values:literals(json!({"source_id":id,"event_key":event_key,"evidence_hash":evidence_hash,"source_url":url,"observed_at":now,"facts":facts,"review_status":"review"}))}]);
        }
    }
    let Some(event_entity) = store
        .schema
        .entities
        .iter()
        .find(|entity| entity.name == "gig_event")
    else {
        return;
    };
    for event in events {
        let mut mutations = Vec::new();
        let evidence_hash =
            pharos_runtime_database::operations::operation_fingerprint(&event.evidence)
                .unwrap_or_default();
        if evidence_hash.is_empty() {
            continue;
        }
        let mut observation = json!({"source_id":id,"event_key":event.key,"evidence_hash":evidence_hash,"source_url":event.evidence_source,"observed_at":now,"facts":event.evidence});
        if supports_review {
            observation["review_status"] = json!("accepted");
        }
        mutations.push(Mutation::InsertIfAbsent {
            entity: "event_observation".into(),
            unique_key: vec!["source_id".into(), "evidence_hash".into()],
            values: literals(observation),
        });
        let existing = existing_events.get(&event.key);
        if existing
            .as_ref()
            .is_some_and(|row| row.get("manual_locked").and_then(Value::as_bool) == Some(true))
        {
            groups.push(mutations);
            continue;
        }
        let filter = if let Some(existing) = existing {
            Predicate::All(
                existing
                    .iter()
                    .map(|(name, value)| Predicate::Equal(name.clone(), value.clone()))
                    .collect(),
            )
        } else {
            Predicate::Equal("event_key".into(), json!(event.key))
        };
        mutations.push(Mutation::RequireCount {
            entity: "gig_event".into(),
            filter,
            minimum: u64::from(existing.is_some()),
            maximum: Some(u64::from(existing.is_some())),
        });
        let matched_organizers = event
            .organizers
            .iter()
            .filter_map(|name| organizers.get(&normalized_name(name)))
            .collect::<Vec<_>>();
        let Some(mut values) = model::write_values(
            existing,
            &event,
            artist_id,
            id,
            &now,
            matched_organizers
                .first()
                .and_then(|row| row.get("id"))
                .and_then(Value::as_i64),
        ) else {
            return;
        };
        values.retain(|name, _| event_entity.fields.iter().any(|field| field.name == *name));
        mutations.push(Mutation::Upsert {
            entity: "gig_event".into(),
            unique_key: vec!["event_key".into()],
            values: literals(Value::Object(values)),
        });
        mutations.push(Mutation::InsertIfAbsent {
            entity: "gig_artist".into(),
            unique_key: vec!["artist_row_id".into(), "event_key".into()],
            values: literals(json!({"artist_row_id":artist_id,"event_key":event.key})),
        });
        for organizer in matched_organizers {
            mutations.push(Mutation::RequireCount {
                entity: "organizer".into(),
                filter: Predicate::All(
                    organizer
                        .iter()
                        .map(|(name, value)| Predicate::Equal(name.clone(), value.clone()))
                        .collect(),
                ),
                minimum: 1,
                maximum: Some(1),
            });
            mutations.push(Mutation::InsertIfAbsent {
                entity: "gig_organizer".into(),
                unique_key: vec!["event_key".into(), "organizer_id".into()],
                values: literals(json!({"event_key":event.key,"organizer_id":organizer["id"]})),
            });
        }
        groups.push(mutations);
    }
    let mut source_values = json!({"next_check_at":pharos_runtime_core::time::timestamp_after_seconds_utc(if status=="pending" {30} else {SOURCE_CHECK_INTERVAL_SECONDS}),"last_status":status,"last_message":detail,"last_checked_at":now,"enabled":!removed});
    if matches!(
        status,
        "success" | "no_events" | "review" | "reviewed" | "social_collected"
    ) && store.schema.entities.iter().any(|entity| {
        entity.name == "event_source"
            && entity
                .fields
                .iter()
                .any(|field| field.name == "last_success_at")
    }) {
        source_values["last_success_at"] = json!(now);
    }
    let completion = Mutation::Update {
        entity: "event_source".into(),
        filter: Predicate::Equal("id".into(), json!(id)),
        values: literals(source_values),
        expected_rows: Some(1),
    };
    if let Err(error) = commit_event_groups(store, &common_guards, groups, completion) {
        eprintln!("event discovery acceptance failed: {error:?}");
        // A rejected event batch must not leave the source looking indefinitely queued.
        let mut steps = common_guards;
        steps.push(Mutation::Update {entity:"event_source".into(),filter:Predicate::Equal("id".into(),json!(id)),values:literals(json!({"last_status":"failed","last_message":"Event evidence could not be saved; no unvalidated performance was published","last_checked_at":now,"next_check_at":pharos_runtime_core::time::timestamp_after_seconds_utc(SOURCE_CHECK_INTERVAL_SECONDS)})),expected_rows:Some(1)});
        if let Err(problem) = store.adapter.transact(
            &store.config,
            &store.schema,
            &Transaction {
                steps,
                idempotency_key: None,
            },
        ) {
            eprintln!("event discovery failure status unavailable: {problem:?}");
        }
    }
}

/// Commits independent event groups below the shared transaction budget, retaining the lease until completion.
fn commit_event_groups(
    store: &super::database::LogicalAppStore,
    guards: &[Mutation],
    groups: Vec<Vec<Mutation>>,
    completion: Mutation,
) -> Result<(), pharos_runtime_database::operations::DatabaseError> {
    if groups
        .iter()
        .any(|group| group.len() + guards.len() + 1 > 200)
    {
        return Err(pharos_runtime_database::operations::DatabaseError::InvalidInput);
    }
    let mut steps = guards.to_vec();
    for group in groups {
        if steps.len() + group.len() + 1 > 200 {
            store.adapter.transact(
                &store.config,
                &store.schema,
                &Transaction {
                    steps: std::mem::replace(&mut steps, guards.to_vec()),
                    idempotency_key: None,
                },
            )?;
        }
        steps.extend(group);
    }
    steps.push(completion);
    store.adapter.transact(
        &store.config,
        &store.schema,
        &Transaction {
            steps,
            idempotency_key: None,
        },
    )?;
    Ok(())
}

/// Separates validated public-source work from social evidence requiring explicit publication review.
struct SocialIntake {
    link_count: usize,
    unverified_posts: usize,
    verified_posts: usize,
    events: Vec<DiscoveredEvent>,
    reviews: Vec<ReviewCandidate>,
    guards: Vec<Mutation>,
}

/// Keeps only bounded event declarations from inert social evidence; graph metadata and unrelated posts are excluded.
fn social_review_candidates(value: &Value, depth: usize, output: &mut Vec<ReviewCandidate>) {
    let now = runtime_timestamp_seconds(&current_timestamp_utc()).unwrap_or(u64::MAX);
    if depth > 12 || output.len() >= 40 {
        return;
    }
    if value.get("evidence_kind").and_then(Value::as_str) == Some("rendered_link") {
        let event_link = value
            .get("url")
            .and_then(Value::as_str)
            .and_then(public_link)
            .and_then(|value| url::Url::parse(&value).ok())
            .is_some_and(|url| {
                url.path_segments().is_some_and(|mut segments| {
                    segments.any(|segment| {
                        matches!(
                            segment.to_ascii_lowercase().as_str(),
                            "ticket"
                                | "tickets"
                                | "event"
                                | "events"
                                | "festival"
                                | "festivals"
                                | "gig"
                                | "gigs"
                        )
                    })
                })
            });
        if event_link && social_evidence::eligible(value, now) && value.to_string().len() <= 7680 {
            let mut facts = value.clone();
            facts["@type"] = json!("Event");
            output.push(ReviewCandidate {facts,reason:"Event or ticket link observed on the authorized artist page; verify event facts and public publication rights before publishing",accepted_event_id:None});
        }
        return;
    }
    if value.get("evidence_kind").and_then(Value::as_str) == Some("rendered_post") {
        let text = format!(
            "{} {}",
            value
                .get("description")
                .and_then(Value::as_str)
                .unwrap_or(""),
            value
                .get("image_text")
                .and_then(Value::as_str)
                .unwrap_or("")
        )
        .to_lowercase();
        if [
            "event",
            "festival",
            "gig",
            "perform",
            "playing",
            "live set",
            "ticket",
            "concert",
            "tour",
            "出演",
            "イベント",
            "フェス",
            "ライブ",
            "公演",
        ]
        .iter()
        .any(|word| text.contains(word))
            && social_evidence::eligible(value, now)
            && value.to_string().len() <= 7680
        {
            let mut candidate = value.clone();
            candidate["@type"] = json!("Event");
            output.push(ReviewCandidate {facts:candidate,reason:"Rendered social post may describe a performance; verify dates, artist, venue and public publication rights before publishing",accepted_event_id:None});
        }
        return;
    }
    if value.get("@type").is_some_and(|kind| {
        let matches = |kind: &str| {
            matches!(
                kind.rsplit('/').next(),
                Some("Event" | "MusicEvent" | "Festival")
            )
        };
        kind.as_str().is_some_and(matches)
            || kind
                .as_array()
                .is_some_and(|kinds| kinds.iter().filter_map(Value::as_str).any(matches))
    }) && social_evidence::eligible(value, now)
        && value.to_string().len() <= 7680
    {
        output.push(ReviewCandidate {facts:value.clone(),reason:"Authorized social event evidence; verify dates, artist, venue and public publication rights before publishing",accepted_event_id:None});
    }
    if let Some(values) = value
        .as_array()
        .or_else(|| value.get("@graph").and_then(Value::as_array))
    {
        for value in values.iter().take(40) {
            social_review_candidates(value, depth + 1, output);
        }
    }
}

/// Converts authorized post links into public-page work, never interpreting social prose as event facts.
#[cfg(not(target_os = "wasi"))]
fn queue_social_links(
    app: &AppRuntime,
    store: &super::database::LogicalAppStore,
    profile: &str,
    artist: i64,
    source_guard: Predicate,
    artist_revision: Value,
) -> Result<SocialIntake, pharos_runtime_social::SocialError> {
    let observations = super::social_connections::observations_for_profile(
        app,
        profile,
        SOURCE_CHECK_INTERVAL_SECONDS as i64,
    )?;
    let mut reviews = Vec::new();
    let artist_name = query(
        store,
        "artists",
        &["name"],
        Predicate::Equal("csv_row_id".into(), json!(artist)),
    )
    .and_then(|row| row.get("name").and_then(Value::as_str).map(str::to_owned))
    .ok_or(pharos_runtime_social::SocialError::InvalidResponse)?;
    let now = runtime_timestamp_seconds(&current_timestamp_utc()).unwrap_or(u64::MAX);
    let mut events = BTreeMap::new();
    for value in observations.evidence.iter().take(40) {
        if let Some(mut event) = social_announcements::extract(value, &artist_name, now) {
            event.key =
                pharos_runtime_database::operations::operation_fingerprint(&(artist, &event.key))
                    .map_err(|_| pharos_runtime_social::SocialError::InvalidResponse)?;
            events.insert(event.key.clone(), event);
        } else {
            social_review_candidates(value, 0, &mut reviews);
        }
    }
    let mut events = events.into_values().collect::<Vec<_>>();
    social_announcements::retain_existing_identities(store, artist, &mut events)?;
    let guards = observations.guards.clone();
    let unverified_posts = observations.unverified_posts;
    let verified_posts = observations.verified_posts;
    let link_count =
        accept_social_links(store, artist, source_guard, artist_revision, observations)?;
    Ok(SocialIntake {
        link_count,
        unverified_posts,
        verified_posts,
        events,
        reviews,
        guards,
    })
}

/// Commits a retained social link batch under source, artist and connection authority guards.
#[cfg(not(target_os = "wasi"))]
fn accept_social_links(
    store: &super::database::LogicalAppStore,
    artist: i64,
    source_guard: Predicate,
    artist_revision: Value,
    observations: super::social_connections::ObservationBatch,
) -> Result<usize, pharos_runtime_social::SocialError> {
    let fields = std::iter::once("csv_row_id")
        .chain(PROFILE_FIELDS)
        .collect::<Vec<_>>();
    let artist_profile = query(
        store,
        "artists",
        &fields,
        Predicate::Equal("csv_row_id".into(), json!(artist)),
    )
    .ok_or(pharos_runtime_social::SocialError::AuthorizationRequired)?;
    let organizers =
        organizer_sites(store).map_err(|_| pharos_runtime_social::SocialError::Transport)?;
    let mut links = std::collections::BTreeSet::new();
    for url in observations.links {
        let Some(url) = public_link(&url)
            .and_then(|value| url::Url::parse(&value).ok())
            .filter(|url| {
                !requires_social_authorization(url)
                    && eligible_source(url.as_str(), &artist_profile, &organizers)
            })
        else {
            continue;
        };
        if url.as_str().len() > 512 {
            return Err(pharos_runtime_social::SocialError::Capacity);
        }
        links.insert(url.to_string());
    }
    if links.len() > 1000 {
        return Err(pharos_runtime_social::SocialError::Capacity);
    }
    let now = current_timestamp_utc();
    let mut steps = observations.guards;
    steps.push(Mutation::RequireCount {
        entity: "event_source".into(),
        filter: source_guard,
        minimum: 1,
        maximum: Some(1),
    });
    steps.push(Mutation::RequireCount {
        entity: "artists".into(),
        filter: Predicate::All(vec![
            Predicate::Equal("csv_row_id".into(), json!(artist)),
            Predicate::Equal("crm_status".into(), json!("active")),
            Predicate::Equal("record_updated_at".into(), artist_revision),
        ]),
        minimum: 1,
        maximum: Some(1),
    });
    for url in &links {
        steps.push(Mutation::InsertIfAbsent {entity:"event_source".into(),unique_key:vec!["artist_row_id".into(),"url".into()],values:literals(json!({"artist_row_id":artist,"url":url,"enabled":true,"managed":false,"next_check_at":now,"last_status":"pending","last_checked_at":"","last_message":"From authorized social content; public evidence validation pending"}))});
    }
    store
        .adapter
        .transact(
            &store.config,
            &store.schema,
            &Transaction {
                steps,
                idempotency_key: None,
            },
        )
        .map_err(|_| pharos_runtime_social::SocialError::Transport)?;
    Ok(links.len())
}

/// Keeps unsupported host targets explicitly unavailable rather than attempting unprotected page collection.
#[cfg(target_os = "wasi")]
fn queue_social_links(
    _app: &AppRuntime,
    _store: &super::database::LogicalAppStore,
    _profile: &str,
    _artist: i64,
    _source_guard: Predicate,
    _artist_revision: Value,
) -> Result<SocialIntake, pharos_runtime_social::SocialError> {
    Err(pharos_runtime_social::SocialError::UnsupportedProvider)
}

/// Requeues authorization-blocked sources after an accepted connection, retaining source and connection race guards.
pub(super) fn social_connection_available(
    app: &AppRuntime,
    provider: pharos_runtime_social::provider::SocialProvider,
    guards: &[Mutation],
) -> Result<(), pharos_runtime_social::SocialError> {
    use pharos_runtime_social::SocialError;
    if dynamic::dynamic_contract_doc(app, "event_discovery_contract")
        .and_then(|value| value.get("enabled").and_then(Value::as_bool))
        != Some(true)
    {
        return Ok(());
    }
    let store = app_database_operations(app).ok_or(SocialError::Configuration)?;
    let rows = store
        .adapter
        .query(
            &store.config,
            &store.schema,
            &Query {
                entity: "event_source".into(),
                fields: vec![
                    "id".into(),
                    "url".into(),
                    "next_check_at".into(),
                    "last_status".into(),
                ],
                filter: Some(Predicate::All(vec![
                    Predicate::Equal("enabled".into(), json!(true)),
                    Predicate::In(
                        "last_status".into(),
                        vec![
                            json!("authorization_required"),
                            json!("connector_unavailable"),
                        ],
                    ),
                ])),
                order: vec![],
                limit: 1000,
                offset: 0,
            },
        )
        .map_err(|_| SocialError::Transport)?;
    let now = current_timestamp_utc();
    let mut changes = Vec::new();
    for row in rows {
        if row
            .get("url")
            .and_then(Value::as_str)
            .and_then(|value| url::Url::parse(value).ok())
            .as_ref()
            .and_then(pharos_runtime_social::provider::SocialProvider::from_url)
            != Some(provider)
        {
            continue;
        }
        changes.push(Mutation::Update {entity:"event_source".into(),filter:Predicate::All(vec![Predicate::Equal("id".into(),row["id"].clone()),Predicate::Equal("next_check_at".into(),row["next_check_at"].clone()),Predicate::Equal("last_status".into(),row["last_status"].clone()),Predicate::Equal("enabled".into(),json!(true))]),values:literals(json!({"next_check_at":now,"last_status":"pending","last_message":"Website connection saved; collection queued"})),expected_rows:Some(1)});
    }
    for batch in changes.chunks(150) {
        let mut steps = guards.to_vec();
        steps.extend_from_slice(batch);
        store
            .adapter
            .transact(
                &store.config,
                &store.schema,
                &Transaction {
                    steps,
                    idempotency_key: None,
                },
            )
            .map_err(|_| SocialError::Busy)?;
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Never relaxes a source restriction by truncating patterns or ignoring encoded paths and queries.
    #[test]
    fn robots_rules_fail_closed_for_unsupported_patterns() {
        assert!(robots_body_allows(
            "/events",
            "User-agent: *\nDisallow: /private"
        ));
        assert!(!robots_body_allows(
            "/private/events",
            "User-agent: *\nDisallow: /private"
        ));
        assert!(robots_body_allows(
            "/private/public",
            "User-agent: *\nDisallow: /private\nAllow: /private/public"
        ));
        assert!(!robots_body_allows(
            "/private/events",
            "User-agent: *\nAllow: /\nDisallow: /*private"
        ));
        assert!(!robots_body_allows(
            "/%70rivate",
            "User-agent: *\nDisallow: /private"
        ));
        assert!(!robots_body_allows(
            "/?private=1",
            "User-agent: *\nDisallow: /?private"
        ));
        assert!(!robots_body_allows(
            "/events",
            "User-agent: *\nCrawl-delay: 60"
        ));
    }

    /// Imported catalog references and arbitrary post links never expand polling authority.
    #[test]
    fn discovery_sources_stay_on_declared_sites() {
        let artist=json!({"website":"https://artist.example/","facebook":"https://www.facebook.com/artist","source_url":"https://label.example/artists/artist"}).as_object().unwrap().clone();
        let organizers = vec![url::Url::parse("https://festival.example").unwrap()];
        let direct =
            json!({"website":"https://artist.example/","source_url":"https://artist.example/"})
                .as_object()
                .unwrap()
                .clone();
        assert!(eligible_source(
            "https://artist.example/events",
            &direct,
            &[]
        ));
        let social_website = json!({"website":"https://facebook.com/artist"})
            .as_object()
            .unwrap()
            .clone();
        assert!(!eligible_source(
            "https://facebook.com/other",
            &social_website,
            &[]
        ));
        for value in [
            "https://artist.example/events",
            "https://www.artist.example/events",
            "https://festival.example/gig",
            "https://facebook.com/artist/posts/123",
        ] {
            assert!(eligible_source(value, &artist, &organizers), "{value}");
        }
        for value in [
            "https://www.bing.com/search?q=artist",
            "https://linktr.ee/artist",
            "https://label.example/artists/artist?tracking=1",
            "https://unrelated.example/event",
            "https://facebook.com/anotherartist",
            "https://artist.example.evil.test/events",
        ] {
            assert!(!eligible_source(value, &artist, &organizers), "{value}");
        }
    }

    /// Keeps social login requirements explicit without misclassifying lookalike hosts.
    #[test]
    fn social_sources_require_authorized_connectors() {
        for host in ["facebook.com", "www.instagram.com", "twitter.com", "x.com"] {
            assert!(requires_social_authorization(
                &url::Url::parse(&format!("https://{host}/artist")).unwrap()
            ));
        }
        for host in ["artist.example", "notfacebook.com", "facebook.com.example"] {
            assert!(!requires_social_authorization(
                &url::Url::parse(&format!("https://{host}/artist")).unwrap()
            ));
        }
    }

    /// Exercises durable discovery on an isolated SQLite database or an operator-selected backend test namespace.
    #[test]
    #[ignore = "requires PHAROS_CRM_CONTRACT pointing to the staged music catalog contract"]
    fn recorded_events_are_idempotent_and_preserve_manual_corrections() {
        let path = std::env::var("PHAROS_CRM_CONTRACT").unwrap();
        let value =
            pharos_runtime_compiler::database_yaml::parse(&std::fs::read_to_string(path).unwrap())
                .unwrap();
        let contract: pharos_runtime_database::contract::AppDataContract =
            serde_json::from_value(value.clone()).unwrap();
        let dir = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
            .join("tmp/0.8.4-testing/music-crm/discovery-tests");
        std::fs::create_dir_all(&dir).unwrap();
        let db = dir.join(format!(
            "events-{}-{}.sqlite3",
            std::process::id(),
            super::super::current_timestamp_utc().replace([':', '.'], "-")
        ));
        let backend = std::env::var("PHAROS_CRM_TEST_BACKEND").unwrap_or_else(|_| "sqlite".into());
        let mut config = pharos_runtime_database::DbBackendConfig {
            backend: backend.clone(),
            host: "127.0.0.1".into(),
            tls_mode: "disable".into(),
            ..Default::default()
        };
        if backend != "sqlite" {
            config.database = std::env::var("PHAROS_CRM_TEST_DATABASE").unwrap();
            assert!(
                config.database.starts_with("pharos_music_cert_")
                    && config.database.len() <= 63
                    && config
                        .database
                        .bytes()
                        .all(|byte| byte.is_ascii_alphanumeric() || byte == b'_')
            );
        }
        match backend.as_str() {
            "sqlite" => config.sqlite_path = Some(db.clone()),
            "postgresql" => {
                config.port = "25432".into();
                config.user = std::env::var("PHAROS_CRM_TEST_PG_USER").unwrap();
                config.password_env_name = "PHAROS_DB_PARITY_PG_PASSWORD".into();
            }
            "mysql" => {
                config.port = "23306".into();
                config.user = "root".into();
                config.password_env_name = "PHAROS_TEST_MYSQL_ROOT_PASSWORD".into();
            }
            "mongodb" => {
                config.port = "27018".into();
                config.options = Some("directConnection=true".into());
            }
            _ => panic!("unsupported isolated backend"),
        }
        let adapter = pharos_runtime_database_adapters::resolve(&config).unwrap();
        adapter.ensure_database(&config, &contract.schema).unwrap();
        adapter.apply_contract(&config, &contract).unwrap();
        let store = super::super::database::LogicalAppStore {
            config,
            schema: contract.schema,
            adapter,
        };
        // Imported catalog rows are replayed by their historical migrations. Keep this
        // isolated test focused on its own artist, without fetching imported profiles.
        store
            .adapter
            .transact(
                &store.config,
                &store.schema,
                &Transaction {
                    steps: vec![Mutation::Update {
                        entity: "artists".into(),
                        filter: Predicate::Equal("crm_status".into(), json!("active")),
                        values: literals(json!({"crm_status":"inactive"})),
                        expected_rows: None,
                    }],
                    idempotency_key: None,
                },
            )
            .unwrap();
        let mut fields = Map::new();
        for field in value["schema"]["entities"]
            .as_array()
            .unwrap()
            .iter()
            .find(|e| e["name"] == "artists")
            .unwrap()["fields"]
            .as_array()
            .unwrap()
        {
            let name = field["name"].as_str().unwrap();
            if name == "csv_row_id" || !field["default"].is_null() {
                continue;
            }
            fields.insert(
                name.into(),
                if field["kind"] == "integer" {
                    json!(1)
                } else if field["nullable"] == true {
                    Value::Null
                } else {
                    json!("")
                },
            );
        }
        fields.insert("name".into(), json!("Mad Maxx"));
        fields.insert("website".into(), json!("https://artist.example/events"));
        store
            .adapter
            .transact(
                &store.config,
                &store.schema,
                &Transaction {
                    steps: vec![Mutation::Insert {
                        entity: "artists".into(),
                        values: literals(Value::Object(fields)),
                    }],
                    idempotency_key: None,
                },
            )
            .unwrap();
        let event = json!({"@type":"MusicEvent","name":"Recorded test performance","startDate":pharos_runtime_core::time::timestamp_after_seconds_utc(86400),"location":{"name":"Recorded venue"},"performer":{"name":"Mad Maxx"},"url":"https://events.example/performance","offers":{"url":"https://tickets.example/performance"}});
        let run = |event: &Value| {
            store
                .adapter
                .transact(
                    &store.config,
                    &store.schema,
                    &Transaction {
                        steps: vec![Mutation::Update {
                            entity: "event_source".into(),
                            filter: Predicate::Equal("enabled".into(), json!(true)),
                            values: literals(json!({"next_check_at":"2020-01-01T00:00:00Z"})),
                            expected_rows: None,
                        }],
                        idempotency_key: None,
                    },
                )
                .unwrap();
            process_store_source(&store, |_| {
                Ok(pharos_runtime_integrations::webhook::WebhookResponse {
                    status_code: 200,
                    body: event.to_string(),
                })
            });
        };
        // The first poll discovers the artist URL and accepts the recorded response.
        run(&event);
        run(&event);
        let source_state = query(
            &store,
            "event_source",
            &[
                "last_status",
                "last_message",
                "enabled",
                "url",
                "last_checked_at",
                "next_check_at",
            ],
            Predicate::Equal("id".into(), json!(1)),
        );
        let state = source_state.as_ref().unwrap();
        let checked =
            runtime_timestamp_seconds(state["last_checked_at"].as_str().unwrap()).unwrap();
        let next = runtime_timestamp_seconds(state["next_check_at"].as_str().unwrap()).unwrap();
        assert!(
            (43_100..=43_300).contains(&(next - checked)),
            "discovery must run twice daily"
        );
        let gig = query(
            &store,
            "gig_event",
            &["event_key", "title", "status"],
            Predicate::Equal("title".into(), json!("Recorded test performance")),
        )
        .unwrap_or_else(|| panic!("missing accepted gig; source: {source_state:?}"));
        let count = store
            .adapter
            .query(
                &store.config,
                &store.schema,
                &Query {
                    entity: "gig_event".into(),
                    fields: vec!["id".into()],
                    filter: None,
                    order: vec![],
                    limit: 10,
                    offset: 0,
                },
            )
            .unwrap();
        assert_eq!(count.len(), 1);
        // The same performance discovered through two explicit artist sources keeps one event and both relationships.
        let artist_fields = store
            .schema
            .entities
            .iter()
            .find(|entity| entity.name == "artists")
            .unwrap()
            .fields
            .iter()
            .map(|field| field.name.as_str())
            .collect::<Vec<_>>();
        let mut second = query(
            &store,
            "artists",
            &artist_fields,
            Predicate::All(vec![
                Predicate::Equal("name".into(), json!("Mad Maxx")),
                Predicate::Equal("crm_status".into(), json!("active")),
            ]),
        )
        .unwrap();
        second.remove("csv_row_id");
        second.insert(
            "id".into(),
            if second.get("id").is_some_and(Value::is_string) {
                json!("controlled-secondary-artist")
            } else {
                json!(1000000)
            },
        );
        second.insert("name".into(), json!("Mad Tribe"));
        second.insert(
            "website".into(),
            json!("https://otherartist.example/events"),
        );
        store
            .adapter
            .transact(
                &store.config,
                &store.schema,
                &Transaction {
                    steps: vec![
                        Mutation::Insert {
                            entity: "artists".into(),
                            values: literals(Value::Object(second)),
                        },
                        Mutation::Update {
                            entity: "event_source".into(),
                            filter: Predicate::Equal("id".into(), json!(1)),
                            values: literals(json!({"enabled":false})),
                            expected_rows: Some(1),
                        },
                    ],
                    idempotency_key: None,
                },
            )
            .unwrap();
        let mut shared_event = event.clone();
        shared_event["performer"] = json!([{"name":"Mad Maxx"},{"name":"Mad Tribe"}]);
        for _ in 0..2 {
            process_store_source(&store, |_| {
                Ok(pharos_runtime_integrations::webhook::WebhookResponse {
                    status_code: 200,
                    body: shared_event.to_string(),
                })
            });
        }
        let links = store
            .adapter
            .query(
                &store.config,
                &store.schema,
                &Query {
                    entity: "gig_artist".into(),
                    fields: vec!["artist_row_id".into()],
                    filter: Some(Predicate::Equal(
                        "event_key".into(),
                        gig["event_key"].clone(),
                    )),
                    order: vec![],
                    limit: 10,
                    offset: 0,
                },
            )
            .unwrap();
        assert_eq!(links.len(), 2);
        store
            .adapter
            .transact(
                &store.config,
                &store.schema,
                &Transaction {
                    steps: vec![
                        Mutation::Update {
                            entity: "event_source".into(),
                            filter: Predicate::Equal(
                                "url".into(),
                                json!("https://otherartist.example/events"),
                            ),
                            values: literals(json!({"enabled":false})),
                            expected_rows: Some(1),
                        },
                        Mutation::Update {
                            entity: "event_source".into(),
                            filter: Predicate::Equal("id".into(), json!(1)),
                            values: literals(json!({"enabled":true})),
                            expected_rows: Some(1),
                        },
                    ],
                    idempotency_key: None,
                },
            )
            .unwrap();
        let mut rescheduled = event.clone();
        rescheduled["startDate"] = json!(pharos_runtime_core::time::timestamp_after_seconds_utc(
            172800
        ));
        rescheduled["eventStatus"] = json!("https://schema.org/EventRescheduled");
        run(&rescheduled);
        assert_eq!(
            runtime_timestamp_seconds(
                query(
                    &store,
                    "gig_event",
                    &["starts_at"],
                    Predicate::Equal("event_key".into(), gig["event_key"].clone())
                )
                .unwrap()["starts_at"]
                    .as_str()
                    .unwrap()
            ),
            runtime_timestamp_seconds(rescheduled["startDate"].as_str().unwrap())
        );
        // Failed/forbidden/rate-limited reads must preserve the accepted event and report their actual state.
        let last_success = query(
            &store,
            "event_source",
            &["last_success_at"],
            Predicate::Equal("id".into(), json!(1)),
        )
        .unwrap()["last_success_at"]
            .clone();
        assert!(!last_success.as_str().unwrap().is_empty());
        for (response, status) in [
            (
                Ok(pharos_runtime_integrations::webhook::WebhookResponse {
                    status_code: 200,
                    body: "<html><title>Sign in</title><input type=\"password\"></html>".into(),
                }),
                "authorization_required",
            ),
            (
                Ok(pharos_runtime_integrations::webhook::WebhookResponse {
                    status_code: 429,
                    body: String::new(),
                }),
                "rate_limited",
            ),
            (
                Ok(pharos_runtime_integrations::webhook::WebhookResponse {
                    status_code: 403,
                    body: String::new(),
                }),
                "authorization_required",
            ),
            (
                Ok(pharos_runtime_integrations::webhook::WebhookResponse {
                    status_code: 503,
                    body: String::new(),
                }),
                "failed",
            ),
            (
                Ok(pharos_runtime_integrations::webhook::WebhookResponse {
                    status_code: 200,
                    body: "x".repeat(1024 * 1024 + 1),
                }),
                "failed",
            ),
            (
                Ok(pharos_runtime_integrations::webhook::WebhookResponse {
                    status_code: 404,
                    body: String::new(),
                }),
                "not_found",
            ),
            (
                Ok(pharos_runtime_integrations::webhook::WebhookResponse {
                    status_code: 410,
                    body: String::new(),
                }),
                "not_found",
            ),
            (Err(SourceReadError::AccessBlocked), "access_blocked"),
        ] {
            store
                .adapter
                .transact(
                    &store.config,
                    &store.schema,
                    &Transaction {
                        steps: vec![Mutation::Update {
                            entity: "event_source".into(),
                            filter: Predicate::Equal("id".into(), json!(1)),
                            values: literals(json!({"next_check_at":"2020-01-01T00:00:00Z"})),
                            expected_rows: Some(1),
                        }],
                        idempotency_key: None,
                    },
                )
                .unwrap();
            process_store_source(&store, |_| response.clone());
            assert_eq!(
                query(
                    &store,
                    "event_source",
                    &["last_success_at"],
                    Predicate::Equal("id".into(), json!(1))
                )
                .unwrap()["last_success_at"],
                last_success
            );
            assert_eq!(
                query(
                    &store,
                    "event_source",
                    &["last_status"],
                    Predicate::Equal("id".into(), json!(1))
                )
                .unwrap()["last_status"],
                status
            );
            assert_eq!(
                query(
                    &store,
                    "gig_event",
                    &["title"],
                    Predicate::Equal("event_key".into(), gig["event_key"].clone())
                )
                .unwrap()["title"],
                event["name"]
            );
        }
        // A reopened store sees the durable accepted event and retained schedule.
        let reopened = super::super::database::LogicalAppStore {
            adapter: pharos_runtime_database_adapters::resolve(&store.config).unwrap(),
            config: store.config.clone(),
            schema: store.schema.clone(),
        };
        assert!(
            query(
                &reopened,
                "gig_event",
                &["event_key"],
                Predicate::Equal("event_key".into(), gig["event_key"].clone())
            )
            .is_some()
        );
        let mut cancelled = event.clone();
        cancelled["eventStatus"] = json!("https://schema.org/EventCancelled");
        run(&cancelled);
        assert_eq!(
            query(
                &store,
                "gig_event",
                &["status"],
                Predicate::Equal("event_key".into(), gig["event_key"].clone())
            )
            .unwrap()["status"],
            "cancelled"
        );
        store
            .adapter
            .transact(
                &store.config,
                &store.schema,
                &Transaction {
                    steps: vec![Mutation::Update {
                        entity: "gig_event".into(),
                        filter: Predicate::Equal("event_key".into(), gig["event_key"].clone()),
                        values: literals(json!({"manual_locked":true,"title":"Owner correction"})),
                        expected_rows: Some(1),
                    }],
                    idempotency_key: None,
                },
            )
            .unwrap();
        run(&event);
        assert_eq!(
            query(
                &store,
                "gig_event",
                &["title"],
                Predicate::Equal("event_key".into(), gig["event_key"].clone())
            )
            .unwrap()["title"],
            "Owner correction"
        );
        // Old, retrospective and unattributed social reviews are removed without touching the accepted gig.
        for (key, candidate, status) in [
            (
                "old-post",
                json!({"evidence_kind":"rendered_post","datePublished":"Dec 30 2024","description":"Playing October 30 2026"}),
                "review",
            ),
            (
                "retrospective",
                json!({"evidence_kind":"rendered_post","datePublished":current_timestamp_utc(),"description":"Thank you for the festival"}),
                "review",
            ),
            (
                "stream",
                json!({"source_profile":"https://facebook.com/artist","description":"private-stream-text"}),
                "review",
            ),
            (
                "owner-decision",
                json!({"evidence_kind":"rendered_post","description":"owner-rejected-evidence"}),
                "rejected",
            ),
        ] {
            store.adapter.transact(&store.config,&store.schema,&Transaction {steps:vec![Mutation::Insert{entity:"event_observation".into(),values:literals(json!({"source_id":1,"event_key":key,"evidence_hash":key,"source_url":"https://facebook.com/artist","observed_at":current_timestamp_utc(),"facts":json!({"candidate":candidate}).to_string(),"review_status":status}))}],idempotency_key:None}).unwrap();
        }
        prune_social_reviews(
            &store,
            runtime_timestamp_seconds(&current_timestamp_utc()).unwrap(),
        )
        .unwrap();
        for key in ["old-post", "retrospective", "stream"] {
            let row = query(
                &store,
                "event_observation",
                &["facts", "review_status"],
                Predicate::Equal("event_key".into(), json!(key)),
            )
            .unwrap();
            assert_eq!(row["review_status"], "rejected");
            assert!(
                !row["facts"]
                    .as_str()
                    .unwrap()
                    .contains("private-stream-text")
            );
        }
        assert!(
            query(
                &store,
                "event_observation",
                &["facts"],
                Predicate::Equal("event_key".into(), json!("owner-decision"))
            )
            .unwrap()["facts"]
                .as_str()
                .unwrap()
                .contains("owner-rejected-evidence")
        );
        prune_social_reviews(
            &store,
            runtime_timestamp_seconds(&current_timestamp_utc()).unwrap(),
        )
        .unwrap();
        // Incomplete structured candidates remain private, idempotent and durably rejected across repeat discovery.
        assert!(
            store
                .schema
                .entities
                .iter()
                .any(|entity| entity.name == "event_observation"
                    && entity
                        .fields
                        .iter()
                        .any(|field| field.name == "review_status")),
            "review schema required for this acceptance case"
        );
        let mut incomplete = event.clone();
        incomplete.as_object_mut().unwrap().remove("startDate");
        run(&incomplete);
        let candidate = query(
            &store,
            "event_observation",
            &["id", "facts", "review_status"],
            Predicate::Equal("review_status".into(), json!("review")),
        )
        .unwrap();
        assert!(
            candidate["facts"]
                .as_str()
                .unwrap()
                .contains("extraction_version")
        );
        store
            .adapter
            .transact(
                &store.config,
                &store.schema,
                &Transaction {
                    steps: vec![Mutation::Update {
                        entity: "event_observation".into(),
                        filter: Predicate::Equal("id".into(), candidate["id"].clone()),
                        values: literals(json!({"review_status":"rejected"})),
                        expected_rows: Some(1),
                    }],
                    idempotency_key: None,
                },
            )
            .unwrap();
        run(&incomplete);
        assert_eq!(
            query(
                &store,
                "event_observation",
                &["review_status"],
                Predicate::Equal("id".into(), candidate["id"].clone())
            )
            .unwrap()["review_status"],
            "rejected"
        );
        assert_eq!(
            query(
                &store,
                "event_source",
                &["last_status"],
                Predicate::Equal("id".into(), json!(1))
            )
            .unwrap()["last_status"],
            "reviewed"
        );
        assert!(
            query(
                &store,
                "event_observation",
                &["id"],
                Predicate::Equal("review_status".into(), json!("review"))
            )
            .is_none()
        );
        // Social links use the same public queue and cannot publish unverified post text.
        let source_artist = query(
            &store,
            "event_source",
            &["artist_row_id"],
            Predicate::Equal("id".into(), json!(1)),
        )
        .unwrap();
        let artist = query(
            &store,
            "artists",
            &["csv_row_id", "record_updated_at"],
            Predicate::Equal("csv_row_id".into(), source_artist["artist_row_id"].clone()),
        )
        .unwrap();
        let artist_id = artist["csv_row_id"].as_i64().unwrap();
        let scope = pharos_runtime_social::connection::ConnectionScope::new(
            &store.schema.app_id,
            "controlled_social",
            7,
        )
        .unwrap();
        let scope_filter = pharos_runtime_social::repository::scope_filter(
            &scope,
            pharos_runtime_social::provider::SocialProvider::X,
        );
        store.adapter.transact(&store.config,&store.schema,&Transaction {steps:vec![Mutation::Insert {entity:"social_connection".into(),values:literals(json!({"application":scope.application(),"namespace":scope.namespace(),"owner":7,"provider":"x","account_id":"123","username":"controlled","registration":"r1","revision":1,"status":"connected","expires":2000000000,"credential_ref":"controlled-reference","lease_token":"","lease_until":0}))}],idempotency_key:None}).unwrap();
        let batch = || super::super::social_connections::ObservationBatch {
            unverified_posts: 0,
            verified_posts: 0,
            evidence: Vec::new(),
            links: vec![
                "https://artist.example/from-social".into(),
                "https://artist.example/from-social".into(),
                "https://x.com/another-profile".into(),
                "https://www.bing.com/search?q=artist".into(),
                "https://linktr.ee/artist".into(),
                "https://unrelated.example/event".into(),
            ],
            guards: vec![Mutation::RequireCount {
                entity: "social_connection".into(),
                filter: Predicate::All(vec![
                    scope_filter.clone(),
                    Predicate::Equal("revision".into(), json!(1)),
                    Predicate::Equal("status".into(), json!("connected")),
                ]),
                minimum: 1,
                maximum: Some(1),
            }],
        };
        let source_guard = Predicate::All(vec![
            Predicate::Equal("id".into(), json!(1)),
            Predicate::Equal("enabled".into(), json!(true)),
        ]);
        for _ in 0..2 {
            assert_eq!(
                accept_social_links(
                    &store,
                    artist_id,
                    source_guard.clone(),
                    artist["record_updated_at"].clone(),
                    batch()
                )
                .unwrap(),
                1
            );
        }
        let public_sources = store
            .adapter
            .query(
                &store.config,
                &store.schema,
                &Query {
                    entity: "event_source".into(),
                    fields: vec!["id".into()],
                    filter: Some(Predicate::Equal(
                        "url".into(),
                        json!("https://artist.example/from-social"),
                    )),
                    order: vec![],
                    limit: 10,
                    offset: 0,
                },
            )
            .unwrap();
        assert_eq!(public_sources.len(), 1);
        store
            .adapter
            .transact(
                &store.config,
                &store.schema,
                &Transaction {
                    steps: vec![Mutation::Update {
                        entity: "social_connection".into(),
                        filter: scope_filter.clone(),
                        values: literals(json!({"status":"disconnected","revision":2})),
                        expected_rows: Some(1),
                    }],
                    idempotency_key: None,
                },
            )
            .unwrap();
        assert!(
            accept_social_links(
                &store,
                artist_id,
                source_guard,
                artist["record_updated_at"].clone(),
                batch()
            )
            .is_err()
        );
        // A full source page must commit in bounded groups, retaining every event and organizer link.
        for name in ["First organizer", "Second organizer"] {
            store
                .adapter
                .transact(
                    &store.config,
                    &store.schema,
                    &Transaction {
                        steps: vec![Mutation::Insert {
                            entity: "organizer".into(),
                            values: literals(json!({"name":name})),
                        }],
                        idempotency_key: None,
                    },
                )
                .unwrap();
        }
        let page = Value::Array(
            (0..100)
                .map(|index| {
                    json!({
                        "@type":"MusicEvent", "@id":format!("https://events.example/batch/{index}"),
                        "name":format!("Batch performance {index}"),
                        "startDate":pharos_runtime_core::time::timestamp_after_seconds_utc(172800),
                        "location":{"name":"Batch venue"},
                        "performer":[{"name":"Mad Maxx"},{"name":"Mad Tribe"}],
                        "organizer":[{"name":"First organizer"},{"name":"Second organizer"}],
                        "url":format!("https://events.example/batch/{index}")
                    })
                })
                .collect(),
        );
        run(&page);
        run(&page);
        let batch_events = store
            .adapter
            .query(
                &store.config,
                &store.schema,
                &Query {
                    entity: "gig_event".into(),
                    fields: vec!["event_key".into()],
                    filter: Some(Predicate::Contains(
                        "title".into(),
                        "Batch performance".into(),
                    )),
                    order: vec![],
                    limit: 1000,
                    offset: 0,
                },
            )
            .unwrap();
        assert_eq!(batch_events.len(), 100);
        let links = store
            .adapter
            .query(
                &store.config,
                &store.schema,
                &Query {
                    entity: "gig_organizer".into(),
                    fields: vec!["event_key".into()],
                    filter: None,
                    order: vec![],
                    limit: 1000,
                    offset: 0,
                },
            )
            .unwrap();
        assert_eq!(links.len(), 200);
        // Paused sources never reach the injected reader.
        store
            .adapter
            .transact(
                &store.config,
                &store.schema,
                &Transaction {
                    steps: vec![Mutation::Update {
                        entity: "event_source".into(),
                        filter: Predicate::Equal("enabled".into(), json!(true)),
                        values: literals(json!({"enabled":false})),
                        expected_rows: Some(2),
                    }],
                    idempotency_key: None,
                },
            )
            .unwrap();
        process_store_source(&store, |_| panic!("paused source fetched"));
        let mut invalid_steps = vec![Mutation::Update {
            entity: "artists".into(),
            filter: Predicate::Equal("csv_row_id".into(), json!(artist_id)),
            values: literals(json!({"source_url":"https://label.example/artists/artist"})),
            expected_rows: Some(1),
        }];
        for url in [
            "https://www.bing.com/search?q=artist",
            "https://linktr.ee/artist",
            "https://label.example/artists/artist",
        ] {
            invalid_steps.push(Mutation::Insert {entity:"event_source".into(),values:literals(json!({"artist_row_id":artist_id,"url":url,"enabled":true,"managed":false,"next_check_at":"2020-01-01T00:00:00Z","last_status":"access_blocked","last_checked_at":"","last_message":""}))});
        }
        store
            .adapter
            .transact(
                &store.config,
                &store.schema,
                &Transaction {
                    steps: invalid_steps,
                    idempotency_key: None,
                },
            )
            .unwrap();
        process_store_source(&store, |_| panic!("unrelated source fetched"));
        let retired = store
            .adapter
            .query(
                &store.config,
                &store.schema,
                &Query {
                    entity: "event_source".into(),
                    fields: vec!["enabled".into()],
                    filter: Some(Predicate::Equal(
                        "last_status".into(),
                        json!("source_removed"),
                    )),
                    order: vec![],
                    limit: 1000,
                    offset: 0,
                },
            )
            .unwrap();
        assert!(retired.len() >= 3);
        assert!(retired.iter().all(|row| row["enabled"] == false));

        // Date-only source evidence must still satisfy the timestamp-typed storage contract.
        let now = current_timestamp_utc();
        let day = pharos_runtime_core::time::timestamp_after_seconds_utc(86400)[..10].to_owned();
        let post = json!({"evidence_kind":"rendered_post","audience":"public","description":format!("Next stop: Schema regression festival {day}"),"datePublished":now,"author_url":"https://www.facebook.com/artist","source_profile":"https://www.facebook.com/artist","url":"https://www.facebook.com/artist/posts/123"});
        let event = social_announcements::extract(
            &post,
            "Mad Maxx",
            runtime_timestamp_seconds(&now).unwrap(),
        )
        .unwrap();
        let mut legacy = event.clone();
        legacy.key =
            pharos_runtime_database::operations::operation_fingerprint(&"legacy-announcement")
                .unwrap();
        legacy.event_url = format!("{}?tracking=legacy", event.event_url);
        let values = model::write_values(None, &legacy, artist_id, 1, &now, None).unwrap();
        for _ in 0..2 {
            store
                .adapter
                .transact(
                    &store.config,
                    &store.schema,
                    &Transaction {
                        steps: vec![Mutation::Upsert {
                            entity: "gig_event".into(),
                            unique_key: vec!["event_key".into()],
                            values: literals(Value::Object(values.clone())),
                        }],
                        idempotency_key: None,
                    },
                )
                .unwrap();
        }
        let mut rediscovered = vec![event];
        social_announcements::retain_existing_identities(&store, artist_id, &mut rediscovered)
            .unwrap();
        let event = rediscovered.remove(0);
        assert_eq!(event.key, legacy.key);
        let rows = store
            .adapter
            .query(
                &store.config,
                &store.schema,
                &Query {
                    entity: "gig_event".into(),
                    fields: vec!["starts_at".into(), "venue".into(), "timezone".into()],
                    filter: Some(Predicate::Equal("event_key".into(), json!(event.key))),
                    order: vec![],
                    limit: 2,
                    offset: 0,
                },
            )
            .unwrap();
        assert_eq!(rows.len(), 1);
        assert!(rows[0]["starts_at"].as_str().unwrap().starts_with(&day));
        assert_eq!(rows[0]["venue"], "");
        assert_eq!(rows[0]["timezone"], "");

        store
            .adapter
            .verify_contract(
                &store.config,
                &pharos_runtime_database::contract::AppDataContract {
                    schema: store.schema.clone(),
                    ..contract
                },
            )
            .unwrap();
        drop(store);
        if db.is_file() {
            std::fs::remove_file(db).unwrap();
        }
    }

    /// Keeps model-free extraction limited to complete future performances for the exact artist.
    #[test]
    /// Keeps source identities stable, rejects contradictory evidence and preserves cancellations.
    fn event_identity_conflicts_and_cancellation_are_guarded() {
        let now = runtime_timestamp_seconds("2026-10-20T00:00:00Z").unwrap();
        let mut input = json!({"@type":"MusicEvent","@id":"https://events.example/id/1","name":"Night","startDate":"2027-01-01T20:00:00+01:00","location":{"name":"Hall","address":{"streetAddress":"Main 1","addressCountry":"France"}},"performer":{"name":"Mad Maxx"},"organizer":{"name":"Local Promoter"},"url":"https://events.example/night"});
        let parse = |value: &Value| {
            extract_events(
                &value.to_string(),
                "Mad Maxx",
                "https://artist.example/events",
                now,
            )
            .remove(0)
        };
        let first = parse(&input);
        assert_eq!(first.country, "France");
        assert_eq!(first.address, "Main 1");
        assert_eq!(first.timezone, "+01:00");
        assert_eq!(first.organizers, ["Local Promoter"]);
        let accepted =
            model::write_values(None, &first, 1, 10, "2026-10-20T00:00:00Z", None).unwrap();
        input["startDate"] = json!("2027-01-02T20:00:00+01:00");
        input["url"] = json!("https://events.example/new-url");
        let changed = parse(&input);
        assert_eq!(first.key, changed.key);
        assert!(model::conflicts(&accepted, &changed, 11));
        assert!(!model::conflicts(&accepted, &changed, 10));
        input["@id"] = json!("https://events.example/id/2");
        assert_ne!(first.key, parse(&input).key);
        let mut cancelled = accepted;
        cancelled.insert("status".into(), json!("cancelled"));
        cancelled.insert("event_state".into(), json!("cancelled"));
        assert!(model::conflicts(&cancelled, &changed, 10));
        let retained = model::write_values(
            Some(&cancelled),
            &first,
            2,
            11,
            "2026-10-20T00:00:00Z",
            None,
        )
        .unwrap();
        assert_eq!(retained["status"], "cancelled");
        assert_eq!(retained["artist_row_id"], 1);
        assert_eq!(retained["source_id"], 10);
        assert_ne!(
            retained["record_updated_at"],
            cancelled["record_updated_at"]
        );
        input["endDate"] = json!("2027-01-01T00:00:00Z");
        assert!(
            extract_events(
                &input.to_string(),
                "Mad Maxx",
                "https://artist.example/events",
                now
            )
            .is_empty()
        );
    }

    /// The discovery window includes the last thirty days and excludes older performances without weakening identity checks.
    #[test]
    fn recent_performances_share_verified_event_validation() {
        let now = runtime_timestamp_seconds("2026-10-23T12:00:00Z").unwrap();
        let mut event = json!({"@type":"MusicEvent","name":"Recent set","startDate":"2026-10-01T20:00:00Z","location":{"name":"Hall"},"performer":{"name":"Shota"},"url":"https://artist.example/event"});
        assert!(event_from_value(&event, "Shota", "https://artist.example/", now).is_some());
        event["startDate"] = json!("2026-09-23T11:59:59Z");
        assert!(event_from_value(&event, "Shota", "https://artist.example/", now).is_none());
        event["startDate"] = json!("2026-09-23T12:00:00Z");
        assert!(event_from_value(&event, "Shota", "https://artist.example/", now).is_some());
        assert!(
            event_from_value(&event, "Another artist", "https://artist.example/", now).is_none()
        );
    }

    #[test]
    fn structured_events_require_verified_identity_time_and_venue() {
        let event = json!({"@type":"MusicEvent","name":"Night","startDate":"2027-01-01T20:00:00+01:00","location":{"name":"Hall"},"performer":{"name":"Mad Maxx"},"url":"https://events.example/night","offers":{"url":"https://tickets.example/night"}});
        let now = runtime_timestamp_seconds("2026-10-20T00:00:00Z").unwrap();
        let events = extract_events(
            &event.to_string(),
            "MAD MAXX",
            "https://artist.example/events",
            now,
        );
        assert_eq!(events.len(), 1);
        assert_eq!(
            extract_events(
                &format!("<SCRIPT TYPE='application/ld+json'>{event}</SCRIPT>"),
                "Mad Maxx",
                "https://artist.example/events",
                now
            )
            .len(),
            1
        );
        assert!(events[0].starts_at.starts_with("2027-01-01T19:00:00"));
        assert!(
            extract_events(
                &event.to_string(),
                "Max",
                "https://artist.example/events",
                now
            )
            .is_empty()
        );
        let mut incomplete = event.clone();
        incomplete["startDate"] = json!("2027-01-01");
        assert!(
            extract_events(
                &incomplete.to_string(),
                "Mad Maxx",
                "https://artist.example/events",
                now
            )
            .is_empty()
        );
        assert!(public_link("javascript:alert(1)").is_none());
        assert!(public_link("https://user:secret@example.com/").is_none());
    }
}