pharos_native/runtime/

social_connections.rs

//! Framework-owned external-account controllers. These routes never issue a local login session.

use super::*;
use pharos_runtime_oauth::{CallbackParameters, Cancellation, SecretValue, Url, wire};
use pharos_runtime_security::credentials::EncryptedFileCredentialBackend;
use pharos_runtime_social::{
    SocialError,
    connection::{ConnectionScope, PendingConnection, SocialClient},
    provider::SocialProvider,
    repository::{ConnectionRecord, ConnectionRepository},
};
use serde::{Deserialize, Serialize};
use serde_json::json;

const PREFIX: &str = "/__pharos/social/";
#[cfg(unix)]
mod browser;
mod consent_ui;
mod settings;

static OPERATOR_CONFIG: std::sync::RwLock<Option<String>> = std::sync::RwLock::new(None);

/// Retains only the trusted serving configuration path so isolated previews can find operator registrations.
pub(super) fn install_operator_config(config: &ServeConfig) -> Result<(), &'static str> {
    *OPERATOR_CONFIG
        .write()
        .map_err(|_| "social operator configuration lock failed")? = config.conf_path.clone();
    Ok(())
}

#[derive(Deserialize, Serialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
struct Registration {
    application: String,
    namespace: String,
    provider: String,
    client_id: String,
    callback: String,
    revision: String,
    client_secret_env: Option<String>,
    #[serde(default)]
    client_secret_ref: Option<String>,
    #[serde(default)]
    instagram_account_id: Option<String>,
}

struct Pending {
    attempt: Option<PendingConnection>,
    scope: ConnectionScope,
    session: String,
    previous: Option<ConnectionRecord>,
    created: std::time::Instant,
    provider: SocialProvider,
}

/// Keeps one-use consent state out of persistence, exports and logs; restart safely invalidates it.
fn pending_registry() -> &'static std::sync::Mutex<std::collections::BTreeMap<String, Pending>> {
    static PENDING: std::sync::OnceLock<
        std::sync::Mutex<std::collections::BTreeMap<String, Pending>>,
    > = std::sync::OnceLock::new();
    PENDING.get_or_init(|| std::sync::Mutex::new(std::collections::BTreeMap::new()))
}

/// Consumes consent only in its initiating scope/session; another actor cannot invalidate it.
fn take_pending(
    key: &str,
    scope: &ConnectionScope,
    provider: SocialProvider,
    session: &str,
) -> Result<Pending, SocialError> {
    let mut registry = pending_registry().lock().map_err(|_| SocialError::Busy)?;
    let pending = registry
        .get(key)
        .ok_or(SocialError::AuthorizationRequired)?;
    if &pending.scope != scope
        || pending.provider != provider
        || pending.session != session
        || pending.created.elapsed() >= Duration::from_secs(300)
    {
        return Err(SocialError::Forbidden);
    }
    if pending.attempt.is_none() {
        return Err(SocialError::AuthorizationRequired);
    }
    let pending = registry
        .remove(key)
        .ok_or(SocialError::AuthorizationRequired)?;
    registry.insert(
        key.into(),
        Pending {
            attempt: None,
            scope: pending.scope.clone(),
            session: pending.session.clone(),
            previous: None,
            created: pending.created,
            provider: pending.provider,
        },
    );
    Ok(pending)
}

/// Loads operator-only registrations from external protected storage, never from generated source.
fn registration(
    app: &AppRuntime,
    scope: &ConnectionScope,
    provider: SocialProvider,
) -> Result<Option<Registration>, SocialError> {
    if let Some(value) = settings::load(app, scope, provider)? {
        return Ok(Some(value));
    }
    let operator_conf = if app.is_editor_preview {
        OPERATOR_CONFIG
            .read()
            .map_err(|_| SocialError::Configuration)?
            .clone()
    } else {
        app.runtime_conf_path.clone()
    };
    let Some(conf) = operator_conf.as_deref() else {
        return Ok(None);
    };
    let Some(path) =
        config::dynamic_config_lookup_raw(Path::new(conf), "SOCIAL_CONNECTIONS_CONFIG")
            .filter(|path| !path.is_empty())
    else {
        return Ok(None);
    };
    let path = config::resolve_conf_relative_path(Path::new(conf), &path)
        .ok_or(SocialError::Configuration)?;
    let metadata = stdfs::symlink_metadata(&path).map_err(|_| SocialError::Configuration)?;
    if !metadata.is_file() || metadata.file_type().is_symlink() || metadata.len() > 65536 {
        return Err(SocialError::Configuration);
    }
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        if metadata.permissions().mode() & 0o077 != 0 {
            return Err(SocialError::Configuration);
        }
    }
    let bytes = stdfs::read(&path).map_err(|_| SocialError::Configuration)?;
    let registrations: Vec<Registration> =
        serde_json::from_slice(&bytes).map_err(|_| SocialError::Configuration)?;
    if registrations.len() > 32 {
        return Err(SocialError::Configuration);
    }
    let mut selected = registrations.into_iter().filter(|row| {
        row.application == scope.application()
            && row.namespace == scope.namespace()
            && row.provider == provider.key()
    });
    let value = selected.next();
    if selected.next().is_some() {
        return Err(SocialError::Configuration);
    }
    Ok(value)
}

/// Freezes an operator registration while leaving protocol endpoints and permissions provider-owned.
fn client(app: &AppRuntime, registration: &Registration) -> Result<SocialClient, SocialError> {
    let secret = registration
        .client_secret_env
        .as_ref()
        .map(|name| {
            if name.is_empty()
                || name.len() > 128
                || !name
                    .bytes()
                    .all(|byte| byte.is_ascii_uppercase() || byte.is_ascii_digit() || byte == b'_')
            {
                return Err(SocialError::Configuration);
            }
            SecretValue::new(std::env::var(name).map_err(|_| SocialError::Configuration)?)
                .map_err(|_| SocialError::Configuration)
        })
        .transpose()?;
    let stored = registration
        .client_secret_ref
        .as_ref()
        .map(|key| {
            use pharos_runtime_security::credentials::CredentialBackend;
            let bytes = zeroize::Zeroizing::new(
                vault(app)?
                    .get_secret(key)
                    .map_err(|_| SocialError::Configuration)?,
            );
            SecretValue::new(
                std::str::from_utf8(&bytes)
                    .map_err(|_| SocialError::Configuration)?
                    .to_owned(),
            )
            .map_err(|_| SocialError::Configuration)
        })
        .transpose()?;
    let mut client = SocialClient::new(
        SocialProvider::parse(&registration.provider)?,
        &registration.client_id,
        &registration.callback,
        &registration.revision,
        stored.as_ref().or(secret.as_ref()),
    )?;
    if let Some(id) = &registration.instagram_account_id {
        client = client.with_instagram_account(id)?;
    }
    Ok(client)
}

/// Restores the verified Instagram account for collection while leaving interactive consent free to select an account.
fn collection_client(
    app: &AppRuntime,
    registration: &Registration,
    scope: &ConnectionScope,
) -> Result<SocialClient, SocialError> {
    let client = client(app, registration)?;
    if client.provider() != SocialProvider::Instagram {
        return Ok(client);
    }
    let store = app_database_operations(app).ok_or(SocialError::Configuration)?;
    let vault = vault(app)?;
    let repository = ConnectionRepository {
        adapter: store.adapter.as_ref(),
        config: &store.config,
        schema: &store.schema,
        vault: &vault,
        guards: &[],
    };
    let row = repository
        .load(scope, SocialProvider::Instagram)?
        .ok_or(SocialError::AuthorizationRequired)?;
    if registration
        .instagram_account_id
        .as_ref()
        .is_some_and(|id| id != &row.account_id)
    {
        return Err(SocialError::Forbidden);
    }
    client.with_instagram_account(&row.account_id)
}

/// Combines storage authority with a host-owned deployment epoch; app names cannot revive old preview grants.
fn connection_namespace(app: &AppRuntime, authority: &str) -> Result<String, SocialError> {
    let epoch = match app.connection_namespace.as_deref() {
        Some(epoch) => epoch,
        None if app.is_editor_preview => return Err(SocialError::Configuration),
        None => return Ok(authority.to_owned()),
    };
    ConnectionScope::new("social", epoch, 1)?;
    let mut digest = Sha256::new();
    digest.update(b"pharos.social.namespace.v1\0");
    digest.update(authority.as_bytes());
    digest.update(b"\0");
    digest.update(epoch.as_bytes());
    Ok(dynamic::hex_digest(&digest.finalize()))
}

/// Derives app/deployment scope exclusively from authenticated storage and host deployment identity.
pub(super) fn scope(app: &AppRuntime, owner: i64) -> Result<ConnectionScope, SocialError> {
    let authority = app_security_cache_scope(app).map_err(|_| SocialError::Configuration)?;
    ConnectionScope::new(
        app.app_id.as_deref().ok_or(SocialError::Configuration)?,
        &connection_namespace(app, &authority)?,
        owner,
    )
}

/// Resolves an external vault beside runtime state, refusing app installation footprints.
fn vault(app: &AppRuntime) -> Result<EncryptedFileCredentialBackend, SocialError> {
    let parent = app
        .state_path
        .as_deref()
        .and_then(|path| Path::new(path).parent())
        .ok_or(SocialError::Configuration)?;
    for root in [app.app_root.as_deref(), app.artifact_dir.as_deref()]
        .into_iter()
        .flatten()
    {
        if parent.starts_with(root) {
            return Err(SocialError::Configuration);
        }
    }
    let directory = parent.join("vault");
    EncryptedFileCredentialBackend::open_or_create(
        directory.join("social-connections.enc"),
        &directory.join("social-root.key"),
    )
    .map_err(|_| SocialError::Transport)
}

/// Pins each request to the selected provider's exact reviewed protocol endpoints and one resource URL.
fn provider_transport(
    client: &SocialClient,
    endpoint: Option<String>,
) -> Result<pharos_runtime_integrations::oauth_http::NativeProviderHttp, SocialError> {
    let mut endpoints = client.protocol_endpoints();
    if let Some(endpoint) = endpoint {
        if client.provider() == SocialProvider::Facebook {
            let url = Url::parse(&endpoint).map_err(|_| SocialError::Configuration)?;
            if let Some(target) = url
                .path()
                .strip_prefix("/v25.0/")
                .and_then(|path| path.strip_suffix("/posts"))
            {
                endpoints.push((wire::Method::GET, client.page_token_endpoint(target)?));
            }
        }
        endpoints.push((wire::Method::GET, endpoint));
    }
    pharos_runtime_integrations::oauth_http::NativeProviderHttp::for_endpoints(
        &endpoints,
        512 * 1024,
    )
    .map_err(|_| SocialError::Configuration)
}

/// Rechecks the actual durable session after network operations, including logout or owner deactivation.
fn require_session(
    store: &database::LogicalAppStore,
    session: &str,
    owner: i64,
) -> Result<(), SocialError> {
    let active = store
        .adapter
        .active_session(
            &store.config,
            &store.schema,
            session,
            SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .map_err(|_| SocialError::Configuration)?
                .as_secs(),
        )
        .map_err(|_| SocialError::Transport)?;
    if active.is_none_or(|session| session.user_id != owner) {
        return Err(SocialError::AuthorizationRequired);
    }
    let users = store
        .adapter
        .query(
            &store.config,
            &store.schema,
            &pharos_runtime_database::operations::Query {
                entity: "user_account".into(),
                fields: vec!["id".into()],
                filter: Some(pharos_runtime_database::operations::Predicate::All(vec![
                    pharos_runtime_database::operations::Predicate::Equal(
                        "id".into(),
                        json!(owner),
                    ),
                    pharos_runtime_database::operations::Predicate::Equal(
                        "is_active".into(),
                        json!(true),
                    ),
                ])),
                order: vec![],
                limit: 1,
                offset: 0,
            },
        )
        .map_err(|_| SocialError::Transport)?;
    if users.is_empty() {
        return Err(SocialError::AuthorizationRequired);
    }
    Ok(())
}

/// Freezes one actual capability grant plus active-user/session assertions into the eventual database write.
fn authority_guards(
    app: &AppRuntime,
    store: &database::LogicalAppStore,
    owner: i64,
    capability: &str,
    session: Option<&str>,
) -> Result<Vec<pharos_runtime_database::operations::Mutation>, SocialError> {
    use pharos_runtime_database::operations::{Mutation, Predicate, Query};
    let mut guards = vec![Mutation::RequireCount {
        entity: "user_account".into(),
        filter: Predicate::All(vec![
            Predicate::Equal("id".into(), json!(owner)),
            Predicate::Equal("is_active".into(), json!(true)),
        ]),
        minimum: 1,
        maximum: Some(1),
    }];
    let ir = app
        .application_ir
        .as_ref()
        .ok_or(SocialError::Configuration)?;
    let roles = ir
        .role_capability_edges
        .iter()
        .filter(|(_, granted)| granted.as_str() == capability)
        .map(|(role, _)| json!(role.as_str()))
        .collect::<Vec<_>>();
    let mut granted = false;
    for (entity, field) in [
        ("user_profile", "account_type"),
        ("store_membership", "role"),
    ] {
        if !store
            .schema
            .entities
            .iter()
            .any(|value| value.name == entity)
        {
            continue;
        }
        let filter = Predicate::All(vec![
            Predicate::Equal("user_id".into(), json!(owner)),
            Predicate::In(field.into(), roles.clone()),
        ]);
        let rows = store
            .adapter
            .query(
                &store.config,
                &store.schema,
                &Query {
                    entity: entity.into(),
                    fields: vec!["user_id".into()],
                    filter: Some(filter.clone()),
                    order: vec![],
                    limit: 1,
                    offset: 0,
                },
            )
            .map_err(|_| SocialError::Transport)?;
        if !rows.is_empty() {
            guards.push(Mutation::RequireCount {
                entity: entity.into(),
                filter,
                minimum: 1,
                maximum: None,
            });
            granted = true;
            break;
        }
    }
    if !granted {
        return Err(SocialError::Forbidden);
    }
    if let Some(session) = session {
        guards.push(Mutation::RequireCount {
            entity: "auth_session".into(),
            filter: Predicate::All(vec![
                Predicate::Equal("user_id".into(), json!(owner)),
                Predicate::Equal("session_token_hash".into(), json!(session)),
                Predicate::IsNull("revoked_at".into()),
                Predicate::GreaterThan("expires_at".into(), json!(current_timestamp_utc())),
            ]),
            minimum: 1,
            maximum: Some(1),
        });
    }
    Ok(guards)
}

/// Routes bounded owner operations before app handlers; capability comes from the app's declared policies.
pub(super) fn response(
    app: &AppRuntime,
    request: &http::HttpRequest,
    relative: &str,
) -> Option<Vec<u8>> {
    response_at(app, request, relative, None)
}

/// Accepts a host-selected callback route for an isolated preview without trusting browser form values.
pub(super) fn response_at(
    app: &AppRuntime,
    request: &http::HttpRequest,
    relative: &str,
    callback_path: Option<&str>,
) -> Option<Vec<u8>> {
    let path = relative.split('?').next().unwrap_or("");
    if !path.starts_with(PREFIX) {
        return None;
    }
    Some(match handle(app, request, relative, path, callback_path) {
        Ok(response) => response,
        Err(error) => failure(error, &request.method),
    })
}

/// Returns only fixed service diagnostics, never provider bodies, callback parameters or token references.
fn failure(error: SocialError, method: &str) -> Vec<u8> {
    let status = match error {
        SocialError::Forbidden => 403,
        SocialError::AuthorizationRequired => 401,
        SocialError::Busy => 409,
        SocialError::RateLimited => 429,
        SocialError::Capacity => 413,
        SocialError::Configuration | SocialError::Transport => 503,
        _ => 422,
    };
    http_json_response(
        status,
        http_reason_phrase(status),
        &json!({"error":error.code(),"message":error.to_string()}).to_string(),
        method,
    )
}

/// Nonsecret website state composed into the same app-owned status renderer as optional API grants.
struct BrowserStatus {
    available: bool,
    pending: bool,
}

/// Renders only a declared app-owned component; the framework supplies escaped nonsecret values.
#[allow(clippy::too_many_arguments)]
fn status_response(
    app: &AppRuntime,
    request: &http::HttpRequest,
    declaration: &Value,
    registered: Option<&Registration>,
    provider: SocialProvider,
    scope: &ConnectionScope,
    row: Option<ConnectionRecord>,
    session: &str,
    callback_path: Option<&str>,
    browser: Option<BrowserStatus>,
) -> Result<Vec<u8>, SocialError> {
    if let Some(template) = declaration.get("status_template").and_then(Value::as_str) {
        if template.len() > 512
            || !template.starts_with("templates/")
            || !template.ends_with(".html")
            || Path::new(template)
                .components()
                .any(|component| !matches!(component, std::path::Component::Normal(_)))
        {
            return Err(SocialError::Configuration);
        }
        let is_browser = browser.is_some();
        let configured = browser
            .as_ref()
            .map_or(registered.is_some(), |state| state.available);
        let connected = row.as_ref().is_some_and(|row| row.status == "connected");
        let saved_browser = is_browser
            && row
                .as_ref()
                .is_some_and(|row| !row.credential_ref.is_empty());
        let pending = browser.as_ref().map_or_else(
            || consent_ui::pending(scope, provider, session),
            |state| state.pending,
        );
        let disconnectable = connected
            || row.as_ref().is_some_and(|row| {
                matches!(row.status.as_str(), "revoking" | "reauthorization_required")
            })
            || is_browser && (saved_browser || pending);
        let feedback = consent_ui::outcome(scope, provider, session);
        let setup_allowed = !is_browser && settings::can_setup(app, declaration, scope.owner())?;
        let callback = if setup_allowed && !configured {
            Some(settings::callback_url(
                app,
                request,
                provider,
                callback_path,
            )?)
        } else {
            None
        };
        let mut params = json!({
                "social_api_hidden":if is_browser{"hidden"}else{""},
                "social_browser_unavailable_hidden":if is_browser && !configured{""}else{"hidden"},
                "social_browser_finish_hidden":"hidden",
                "social_auto_finish_hidden":if is_browser && pending{""}else{"hidden"},
                "social_auto_trigger":if is_browser && pending{"every 2s"}else{"none"},
                "social_disconnect_hidden":if disconnectable {""}else{"hidden"},
                "social_connect_hidden":if pending || disconnectable {"hidden"}else{""},
                "social_secret_required":if provider==SocialProvider::X {""}else{"required"},
                "social_setup_hidden":if setup_allowed && !configured {""}else{"hidden"},
                "social_setup_url":app_scope_path(app,&format!("/__pharos/social/{}/setup",provider.key())),
                "social_callback_url":callback,
                "social_connection_mode":if is_browser{"browser"}else{"oauth"}, "social_extension_hidden":"hidden", "social_browser_hidden":if is_browser{""}else{"hidden"}, "social_browser_failed_hidden":if is_browser && feedback==Some(consent_ui::Outcome::Failed){""}else{"hidden"}, "social_finish_hidden":"hidden", "social_browser_pending_hidden":"hidden", "social_connect_target":"_self", "social_finish_url":app_scope_path(app,&format!("/__pharos/social/{}/finish",provider.key())),
                "social_pending_hidden":if pending{""}else{"hidden"},
                "social_cancelled_hidden":if feedback==Some(consent_ui::Outcome::Cancelled){""}else{"hidden"},
                "social_failed_hidden":if feedback==Some(consent_ui::Outcome::Failed){""}else{"hidden"},
                "social_selection_hidden":if feedback==Some(consent_ui::Outcome::SelectionRequired){""}else{"hidden"},
                "social_facebook_hidden":if provider==SocialProvider::Facebook{""}else{"hidden"},
                "social_instagram_hidden":if provider==SocialProvider::Instagram{""}else{"hidden"},
                "social_x_hidden":if provider==SocialProvider::X{""}else{"hidden"},
                "social_runtime_status_hidden":if configured && !pending{""}else{"hidden"},
                "social_cancel_url":app_scope_path(app,&format!("/__pharos/social/{}/cancel",provider.key())),
                "social_status":row.as_ref().map(|row|row.status.as_str()).unwrap_or("disconnected"),
                "social_username":row.as_ref().map(|row|row.username.as_str()).unwrap_or(""),
                "social_namespace":scope.namespace(),
                "social_provider":provider.key(), "social_provider_label":provider.label(),
                "social_unavailable_hidden":if configured || is_browser {"hidden"}else{""},
                "social_available_hidden":if configured && !is_browser {""}else{"hidden"},
                "social_connect_url":app_scope_path(app,&format!("/__pharos/social/{}/connect",provider.key())),
                "social_disconnect_url":app_scope_path(app,&format!("/__pharos/social/{}/disconnect",provider.key())),
                "social_connect_disabled":if configured && !pending {""} else {"disabled"},
                "social_disconnect_disabled":if configured && disconnectable {""} else {"disabled"},
                "csrf_token":dynamic::csrf_token_for_response(request.cookie.as_deref()),
        });
        for (key, candidate) in [
            ("youtube", SocialProvider::Youtube),
            ("soundcloud", SocialProvider::Soundcloud),
            ("spotify", SocialProvider::Spotify),
        ] {
            params[format!("social_{key}_hidden")] =
                json!(if provider == candidate { "" } else { "hidden" });
        }
        params["social_reconnect_hidden"] = json!(if is_browser { "" } else { "hidden" });
        params["social_reconnect_disabled"] = json!(if saved_browser && !pending && configured {
            ""
        } else {
            "disabled"
        });
        return Ok(app_fragment_response(
            app,
            template,
            params,
            &request.method,
        ));
    }
    Ok(http_json_response(200,"OK",&json!({"provider":provider.key(),"namespace":scope.namespace(),"configured":registered.is_some(),"connection":row}).to_string(),&request.method))
}

/// Accepts only bounded same-app return paths, before any external grant or credential is changed.
fn return_destination(declaration: &Value) -> Result<&str, SocialError> {
    let destination = declaration
        .get("return_path")
        .and_then(Value::as_str)
        .unwrap_or("/__pharos/social/x/status");
    if destination.len() > 2048
        || !destination.starts_with('/')
        || destination.starts_with("//")
        || destination.contains(['\\', '?', '#', ' '])
        || destination.chars().any(char::is_control)
    {
        return Err(SocialError::Configuration);
    }
    Ok(destination)
}

/// Accepts only declared API or website transport; malformed modes never fall back to another authority.
fn browser_mode(declaration: &Value) -> Result<bool, SocialError> {
    match declaration.get("connection_mode") {
        None => Ok(false),
        Some(Value::String(mode)) if mode == "api" => Ok(false),
        Some(Value::String(mode)) if mode == "browser" => Ok(true),
        _ => Err(SocialError::Configuration),
    }
}

/// Applies owner/session/CSRF gates and dispatches the declared external-account transport.
fn handle(
    app: &AppRuntime,
    request: &http::HttpRequest,
    relative: &str,
    path: &str,
    callback_path: Option<&str>,
) -> Result<Vec<u8>, SocialError> {
    let declaration = dynamic::dynamic_contract_doc(app, "social_connections_contract")
        .ok_or(SocialError::Forbidden)?;
    if declaration.get("enabled").and_then(Value::as_bool) != Some(true) {
        return Err(SocialError::Forbidden);
    }
    let owner = active_session_user_id(app, request.cookie.as_deref())
        .map_err(|_| SocialError::Transport)?;
    if owner <= 0 {
        return Err(SocialError::AuthorizationRequired);
    }
    let capability = declaration
        .get("owner_capability")
        .and_then(Value::as_str)
        .filter(|value| {
            !matches!(
                *value,
                "public" | "customer" | "account_login" | "account_logout" | "authenticated_user"
            )
        })
        .ok_or(SocialError::Configuration)?;
    if !actor_has_route_policy(app, owner, capability).map_err(|_| SocialError::Transport)? {
        return Err(SocialError::Forbidden);
    }
    let (provider, operation) = path
        .strip_prefix(PREFIX)
        .and_then(|p| p.split_once('/'))
        .ok_or(SocialError::Configuration)?;
    let provider = SocialProvider::parse(provider)?;
    if !matches!(
        (request.method.as_str(), operation),
        ("GET", "status" | "callback")
            | (
                "POST",
                "connect" | "cancel" | "disconnect" | "setup" | "finish"
            )
    ) {
        return Err(SocialError::Forbidden);
    }
    if request.method == "POST"
        && (!trusted_browser_origin(request)
            || !dynamic::csrf_request_valid(
                request,
                dynamic::request_body_text(request).as_deref(),
            ))
    {
        return Err(SocialError::Forbidden);
    }
    let scope = scope(app, owner)?;
    let store = app_database_operations(app).ok_or(SocialError::Configuration)?;
    let session = request
        .cookie
        .as_deref()
        .and_then(|cookie| dynamic::dynamic_cookie_value(cookie, "pharos_session"))
        .and_then(|token| native_session_token_hash(app, &token))
        .ok_or(SocialError::AuthorizationRequired)?;
    require_session(&store, &session, owner)?;
    if browser_mode(&declaration)? {
        #[cfg(unix)]
        return browser::handle(
            app,
            request,
            &declaration,
            provider,
            operation,
            &scope,
            &store,
            &session,
            capability,
        );
        #[cfg(not(unix))]
        return Err(SocialError::UnsupportedProvider);
    }
    if operation == "finish" {
        return Err(SocialError::Forbidden);
    }
    if operation == "setup" {
        pharos_runtime_security::local_role_cache_revoke_actor(scope.namespace(), owner);
        if !settings::can_setup(app, &declaration, owner)? {
            return Err(SocialError::Forbidden);
        }
        if request.body_len > 16384
            || request.body.len() > 16384
            || request.body_file_path.is_some()
        {
            return Err(SocialError::Capacity);
        }
        let client_id = dynamic::request_form_value(request, "client_id").unwrap_or_default();
        let secret = zeroize::Zeroizing::new(
            dynamic::request_form_value(request, "client_secret").unwrap_or_default(),
        );
        let callback = settings::callback_url(app, request, provider, callback_path)?;
        let destination = return_destination(&declaration)?;
        settings::save(app, &scope, provider, client_id.trim(), &secret, &callback)?;
        return Ok(http_route_redirect_for_request(
            request,
            &app_scope_path(app, destination),
        ));
    }
    if operation == "cancel" {
        consent_ui::cancel(&scope, provider, &session)?;
        return Ok(http_route_redirect_for_request(
            request,
            &app_scope_path(app, return_destination(&declaration)?),
        ));
    }
    let registered = registration(app, &scope, provider)?;
    if operation == "status" {
        let row = if store
            .schema
            .entities
            .iter()
            .any(|entity| entity.name == "social_connection")
        {
            let vault = vault(app)?;
            ConnectionRepository {
                adapter: store.adapter.as_ref(),
                config: &store.config,
                schema: &store.schema,
                vault: &vault,
                guards: &[],
            }
            .load(&scope, provider)?
        } else {
            None
        };
        if relative
            .split_once('?')
            .and_then(|(_, query)| json_commands::request_form_value(query, "format"))
            .as_deref()
            == Some("json")
        {
            return Ok(http_json_response(200,"OK",&json!({"provider":provider.key(),"namespace":scope.namespace(),"mode":"oauth","setup_scope":"application_deployment","connection_scope":"owner","setup_allowed":settings::can_setup(app,&declaration,owner)?,"configured":registered.is_some(),"authorization_pending":consent_ui::pending(&scope,provider,&session),"connection":row}).to_string(),&request.method));
        }
        return status_response(
            app,
            request,
            &declaration,
            registered.as_ref(),
            provider,
            &scope,
            row,
            &session,
            callback_path,
            None,
        );
    }
    let destination = return_destination(&declaration)?;
    let registered = registered.ok_or(SocialError::Configuration)?;
    let client = client(app, &registered)?;
    let vault = vault(app)?;
    let guards = authority_guards(app, &store, owner, capability, Some(&session))?;
    let repository = ConnectionRepository {
        adapter: store.adapter.as_ref(),
        config: &store.config,
        schema: &store.schema,
        vault: &vault,
        guards: &guards,
    };
    if operation == "connect" {
        let pending = client.begin(scope.clone())?;
        let url =
            Url::parse(pending.authorization_url()).map_err(|_| SocialError::Configuration)?;
        let state = url
            .query_pairs()
            .find(|(key, _)| key == "state")
            .ok_or(SocialError::Configuration)?
            .1
            .into_owned();
        let destination = pending.authorization_url().to_owned();
        let previous = repository.load(&scope, provider)?;
        let mut registry = pending_registry().lock().map_err(|_| SocialError::Busy)?;
        registry.retain(|_, pending| {
            pending.created.elapsed() < Duration::from_secs(300)
                && (pending.scope != scope || pending.provider != provider)
        });
        if registry.len() >= 128 {
            return Err(SocialError::Capacity);
        }
        consent_ui::record(&scope, provider, &session, None);
        registry.insert(
            dynamic::hex_digest(&Sha256::digest(state.as_bytes())),
            Pending {
                attempt: Some(pending),
                scope,
                session,
                previous,
                provider,
                created: std::time::Instant::now(),
            },
        );
        return Ok(http_route_redirect_for_request(request, &destination));
    }
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|_| SocialError::Transport)?;
    let http = provider_transport(&client, None)?;
    if operation == "callback" {
        let query = relative
            .split_once('?')
            .map(|(_, query)| query)
            .ok_or(SocialError::AuthorizationRequired)?;
        let callback =
            CallbackParameters::parse(query).map_err(|_| SocialError::AuthorizationRequired)?;
        let state = json_commands::request_form_value(query, "state")
            .ok_or(SocialError::AuthorizationRequired)?;
        let key = dynamic::hex_digest(&Sha256::digest(state.as_bytes()));
        let pending = take_pending(&key, &scope, provider, &session)?;
        let result = runtime.block_on(client.finish(
            &http,
            &Cancellation::default(),
            &scope,
            pending.attempt.ok_or(SocialError::AuthorizationRequired)?,
            callback,
        ));
        let account = match result {
            Ok(account) => account,
            Err(error) => {
                let mut claimed = consent_ui::claim_guard(&key, &scope, provider, &session)?;
                claimed.remove(&key);
                drop(claimed);
                let outcome = if error == SocialError::ResourceSelectionRequired {
                    consent_ui::Outcome::SelectionRequired
                } else if json_commands::request_form_value(query, "error").as_deref()
                    == Some("access_denied")
                {
                    consent_ui::Outcome::Cancelled
                } else {
                    consent_ui::Outcome::Failed
                };
                consent_ui::record(&scope, provider, &session, Some(outcome));
                return Ok(http_route_redirect_for_request(
                    request,
                    &app_scope_path(app, destination),
                ));
            }
        };
        require_session(&store, &session, owner)?;
        pharos_runtime_security::local_role_cache_revoke_actor(scope.namespace(), owner);
        if !actor_has_route_policy(app, owner, capability).map_err(|_| SocialError::Transport)? {
            return Err(SocialError::Forbidden);
        }
        if registration(app, &scope, provider)?.as_ref() != Some(&registered) {
            return Err(SocialError::Forbidden);
        }
        let mut claimed = consent_ui::claim_guard(&key, &scope, provider, &session)?;
        let promoted = repository.promote(
            &client,
            &scope,
            provider,
            &registered.revision,
            pending.previous.as_ref(),
            &account,
        );
        claimed.remove(&key);
        drop(claimed);
        if promoted.is_err() {
            consent_ui::record(
                &scope,
                provider,
                &session,
                Some(consent_ui::Outcome::Failed),
            );
            return Ok(http_route_redirect_for_request(
                request,
                &app_scope_path(app, destination),
            ));
        }
        consent_ui::record(&scope, provider, &session, None);
        let now = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map_err(|_| SocialError::Configuration)?
            .as_secs() as i64;
        repository.resume_collections(&scope, provider, now)?;
    } else {
        let row = repository
            .load(&scope, provider)?
            .ok_or(SocialError::AuthorizationRequired)?;
        let account = if row.status == "revoking" {
            use pharos_runtime_security::credentials::CredentialBackend;
            let bytes = zeroize::Zeroizing::new(
                vault
                    .get_secret(&row.credential_ref)
                    .map_err(|_| SocialError::Transport)?,
            );
            client.restore_from_vault(&scope, &bytes)?
        } else {
            repository.account(&client, &scope, &row, &registered.revision)?
        };
        let revoked = if row.status == "revoking" {
            row
        } else {
            repository.disconnect(&row)?
        };
        repository.cancel_collections(&scope, provider)?;
        runtime.block_on(client.revoke(&http, &Cancellation::default(), &scope, account))?;
        repository.finish_disconnect(&revoked)?;
        consent_ui::record(&scope, provider, &session, None);
        pending_registry()
            .lock()
            .map_err(|_| SocialError::Busy)?
            .retain(|_, pending| pending.scope != scope || pending.provider != provider);
    }
    Ok(http_route_redirect_for_request(
        request,
        &app_scope_path(app, destination),
    ))
}

#[cfg(test)]
mod tests;

/// Verifies retained background authority without requiring the owner to leave a browser session open.
fn collection_context(
    app: &AppRuntime,
    provider: SocialProvider,
) -> Result<(ConnectionScope, database::LogicalAppStore, Registration), SocialError> {
    let (scope, store) = collection_authority(app, None)?;
    let registration =
        registration(app, &scope, provider)?.ok_or(SocialError::AuthorizationRequired)?;
    Ok((scope, store, registration))
}

/// Rechecks durable active-owner authority independently of the selected API or website transport.
fn collection_authority(
    app: &AppRuntime,
    selected_owner: Option<i64>,
) -> Result<(ConnectionScope, database::LogicalAppStore), SocialError> {
    use pharos_runtime_database::operations::{Predicate, Query};
    let declaration = dynamic::dynamic_contract_doc(app, "social_connections_contract")
        .ok_or(SocialError::AuthorizationRequired)?;
    if declaration.get("enabled").and_then(Value::as_bool) != Some(true) {
        return Err(SocialError::AuthorizationRequired);
    }
    let owner = selected_owner
        .or_else(|| declaration.get("collection_owner").and_then(Value::as_i64))
        .ok_or(SocialError::AuthorizationRequired)?;
    let scope = scope(app, owner)?;
    let capability = declaration
        .get("owner_capability")
        .and_then(Value::as_str)
        .filter(|value| {
            !matches!(
                *value,
                "public" | "customer" | "account_login" | "account_logout" | "authenticated_user"
            )
        })
        .ok_or(SocialError::Configuration)?;
    pharos_runtime_security::local_role_cache_revoke_actor(scope.namespace(), owner);
    if !actor_has_route_policy(app, owner, capability).map_err(|_| SocialError::Transport)? {
        return Err(SocialError::Forbidden);
    }
    let store = app_database_operations(app).ok_or(SocialError::Configuration)?;
    let users = store
        .adapter
        .query(
            &store.config,
            &store.schema,
            &Query {
                entity: "user_account".into(),
                fields: vec!["id".into()],
                filter: Some(Predicate::All(vec![
                    Predicate::Equal("id".into(), json!(owner)),
                    Predicate::Equal("is_active".into(), json!(true)),
                ])),
                order: vec![],
                limit: 1,
                offset: 0,
            },
        )
        .map_err(|_| SocialError::Transport)?;
    if users.is_empty() {
        return Err(SocialError::Forbidden);
    }
    Ok((scope, store))
}

/// Refreshes near-expiry authority under the durable lease and immutable revision promotion.
fn current_account(
    repository: &ConnectionRepository<'_>,
    client: &SocialClient,
    scope: &ConnectionScope,
    registration: &Registration,
) -> Result<
    (
        ConnectionRecord,
        pharos_runtime_social::connection::ConnectedAccount,
    ),
    SocialError,
> {
    let row = repository
        .load(scope, client.provider())?
        .ok_or(SocialError::AuthorizationRequired)?;
    let account = repository.account(client, scope, &row, &registration.revision)?;
    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_err(|_| SocialError::Configuration)?
        .as_secs() as i64;
    if row.expires > now + 60 {
        return Ok((row, account));
    }
    let leased = repository.lease_refresh(&row, now)?;
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|_| SocialError::Transport)?;
    let account = runtime.block_on(client.refresh(
        &provider_transport(client, None)?,
        &Cancellation::default(),
        scope,
        &account,
    ))?;
    let updated = repository.promote(
        client,
        scope,
        client.provider(),
        &registration.revision,
        Some(&leased),
        &account,
    )?;
    Ok((updated, account))
}

/// Processes one due social page independently of domain tables, preserving cancellation and restart leases.
pub(super) fn process_one(app: &AppRuntime) {
    #[cfg(unix)]
    browser::maintenance();
    if dynamic::dynamic_contract_doc(app, "social_connections_contract")
        .as_ref()
        .map(browser_mode)
        .transpose()
        .unwrap_or(Some(true))
        .unwrap_or(false)
    {
        return;
    }

    for provider in [
        SocialProvider::Facebook,
        SocialProvider::Instagram,
        SocialProvider::X,
    ] {
        process_provider(app, provider);
    }
}

/// Processes a bounded page independently for each configured provider.
fn process_provider(app: &AppRuntime, provider: SocialProvider) {
    if super::runtime_shutdown_requested() {
        return;
    }
    let Ok((scope, store, registration)) = collection_context(app, provider) else {
        return;
    };
    let Ok(vault) = vault(app) else {
        return;
    };
    let Some(capability) = dynamic::dynamic_contract_doc(app, "social_connections_contract")
        .and_then(|value| {
            value
                .get("owner_capability")
                .and_then(Value::as_str)
                .map(str::to_owned)
        })
    else {
        return;
    };
    let Ok(guards) = authority_guards(app, &store, scope.owner(), &capability, None) else {
        return;
    };
    let repository = ConnectionRepository {
        adapter: store.adapter.as_ref(),
        config: &store.config,
        schema: &store.schema,
        vault: &vault,
        guards: &guards,
    };
    let Ok(client) = collection_client(app, &registration, &scope) else {
        return;
    };
    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|value| value.as_secs() as i64)
        .unwrap_or(0);
    let Ok(Some(lease)) = repository.claim_collection(&scope, provider, now) else {
        return;
    };
    let result = (|| {
        let (row, account) = current_account(&repository, &client, &scope, &registration)?;
        let endpoint =
            client.collection_endpoint(&pharos_runtime_social::connection::PostQuery {
                account_id: &lease.target,
                since_id: (!lease.since_id.is_empty()).then_some(lease.since_id.as_str()),
                page_token: (!lease.page_token.is_empty()).then_some(lease.page_token.as_str()),
            })?;
        let http = provider_transport(&client, Some(endpoint))?;
        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .map_err(|_| SocialError::Transport)?;
        let page = runtime.block_on(repository.collect_page(
            &client,
            &http,
            &Cancellation::default(),
            &scope,
            &account,
            &lease,
        ))?;
        if super::runtime_shutdown_requested() {
            return Err(SocialError::Busy);
        }
        collection_context(app, provider)?;
        repository.commit_page(
            &scope,
            provider,
            &row,
            &lease,
            &page,
            SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .map_err(|_| SocialError::Configuration)?
                .as_secs() as i64,
        )
    })();
    if let Err(error) = result {
        let _ = repository.fail_collection(&lease, error, now);
    }
}

/// Retained inert observations plus guards for atomic acceptance by a domain consumer.
pub(super) struct ObservationBatch {
    pub links: Vec<String>,
    pub evidence: Vec<Value>,
    pub unverified_posts: usize,
    pub verified_posts: usize,
    pub guards: Vec<pharos_runtime_database::operations::Mutation>,
}

/// Resolves an explicit social profile and schedules its retained observations for an authorized consumer.
pub(super) fn observations_for_profile(
    app: &AppRuntime,
    profile: &str,
    interval: i64,
) -> Result<ObservationBatch, SocialError> {
    if dynamic::dynamic_contract_doc(app, "social_connections_contract")
        .as_ref()
        .map(browser_mode)
        .transpose()?
        .unwrap_or(false)
    {
        #[cfg(unix)]
        return browser::observations(app, profile);
        #[cfg(not(unix))]
        return Err(SocialError::UnsupportedProvider);
    }
    let parsed = Url::parse(profile).map_err(|_| SocialError::Configuration)?;
    let provider = SocialProvider::from_url(&parsed).ok_or(SocialError::UnsupportedProvider)?;
    let (scope, store, registration) = collection_context(app, provider)?;
    let client = collection_client(app, &registration, &scope)?;
    let endpoint = client.resource_endpoint(profile)?;
    let vault = vault(app)?;
    let capability = dynamic::dynamic_contract_doc(app, "social_connections_contract")
        .and_then(|value| {
            value
                .get("owner_capability")
                .and_then(Value::as_str)
                .map(str::to_owned)
        })
        .ok_or(SocialError::Configuration)?;
    let guards = authority_guards(app, &store, scope.owner(), &capability, None)?;
    let repository = ConnectionRepository {
        adapter: store.adapter.as_ref(),
        config: &store.config,
        schema: &store.schema,
        vault: &vault,
        guards: &guards,
    };

    let (connection, account) = current_account(&repository, &client, &scope, &registration)?;
    let runtime = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .map_err(|_| SocialError::Transport)?;
    let target = runtime.block_on(client.resolve_profile(
        &provider_transport(&client, Some(endpoint))?,
        &Cancellation::default(),
        &scope,
        &account,
        profile,
    ))?;
    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_err(|_| SocialError::Configuration)?
        .as_secs() as i64;
    repository.schedule(&scope, provider, &target, interval, now)?;
    repository.collection_outcome(&scope, provider, &target)?;
    if !repository.collection_complete(&scope, provider, &target, now)? {
        return Err(SocialError::Busy);
    }
    let mut links = std::collections::BTreeSet::new();
    let mut complete = false;
    for offset in (0..10000).step_by(100) {
        let observations = repository.observations(&scope, provider, &target, offset)?;
        complete = observations.len() < 100;
        for observation in observations {
            let encoded = observation
                .get("links")
                .and_then(Value::as_str)
                .ok_or(SocialError::InvalidResponse)?;
            let urls: Vec<String> =
                serde_json::from_str(encoded).map_err(|_| SocialError::InvalidResponse)?;
            for url in urls {
                links.insert(url);
                if links.len() > 1000 {
                    return Err(SocialError::Capacity);
                }
            }
        }
        if complete {
            break;
        }
    }
    if !complete {
        return Err(SocialError::Capacity);
    }
    let mut guards = guards.clone();
    use pharos_runtime_database::operations::{Mutation, Predicate};
    guards.push(Mutation::RequireCount {
        entity: "social_connection".into(),
        filter: Predicate::All(vec![
            pharos_runtime_social::repository::scope_filter(&scope, provider),
            Predicate::Equal("revision".into(), json!(connection.revision)),
            Predicate::Equal("status".into(), json!("connected")),
        ]),
        minimum: 1,
        maximum: Some(1),
    });
    Ok(ObservationBatch {
        links: links.into_iter().collect(),
        evidence: Vec::new(),
        unverified_posts: 0,
        verified_posts: 0,
        guards,
    })
}

/// Flushes explicitly authorized browser sessions before the native host exits; API grants are unaffected.
#[cfg(unix)]
pub(super) fn shutdown_browsers() {
    browser::shutdown();
}