//! PM-28 authenticated write endpoints.
//!
//! These XRPC routes deliberately use strict OAuth extraction, not the browser
//! redirect extractor. The write order is PDS-first: only after the authenticated
//! actor's PDS accepts a record/blob operation do we eagerly project into SQLite
//! for read-your-own-writes. Hydrant/firehose will later deliver the same event;
//! projection reuse keeps convergence idempotent.

use axum::Json;
use axum::body::Body;
use axum::extract::State;
use axum::http::{HeaderMap, header};
use http_body_util::BodyExt;
use jacquard::client::{Agent, AgentSession, AgentSessionExt};
use jacquard::identity::PublicResolver;
use jacquard::oauth::client::OAuthSession;
use jacquard_axum::oauth::ExtractOAuthSession;
use jacquard_axum::{ExtractXrpc, XrpcResponse};
use jacquard_common::deps::smol_str::SmolStr;
use jacquard_common::types::blob::{BlobRef, MimeType};
use jacquard_common::types::collection::Collection;
use jacquard_common::types::ident::AtIdentifier;
use jacquard_common::types::recordkey::{RecordKey, Rkey};
use jacquard_common::types::string::{AtUri, Cid, Datetime, Did};
use polymodel_api::app_bsky::actor::profile::Profile as BskyProfile;
use polymodel_api::com_atproto::repo::strong_ref::StrongRef;
use polymodel_api::space_polymodel::actor::ProfileView;
use polymodel_api::space_polymodel::actor::bootstrap_profile::{
    BootstrapProfileOutput, BootstrapProfileOutputStatus, BootstrapProfileRequest,
};
use polymodel_api::space_polymodel::actor::profile::Profile as PolymodelProfile;
use polymodel_api::space_polymodel::graph::create_like::{CreateLikeOutput, CreateLikeRequest};
use polymodel_api::space_polymodel::graph::create_save::{CreateSaveOutput, CreateSaveRequest};
use polymodel_api::space_polymodel::graph::create_tag::{CreateTagOutput, CreateTagRequest};
use polymodel_api::space_polymodel::graph::delete_like::{DeleteLikeOutput, DeleteLikeRequest};
use polymodel_api::space_polymodel::graph::delete_save::{DeleteSaveOutput, DeleteSaveRequest};
use polymodel_api::space_polymodel::graph::delete_tag::{DeleteTagOutput, DeleteTagRequest};
use polymodel_api::space_polymodel::graph::like::Like;
use polymodel_api::space_polymodel::graph::save::Save;
use polymodel_api::space_polymodel::graph::tag::Tag;
use polymodel_api::space_polymodel::library::delete_thing::{
    DeleteThingOutput, DeleteThingRequest,
};
use polymodel_api::space_polymodel::library::publish_thing::{
    ModelInput, ModelPublishResult, PublishThingOutput, PublishThingRequest, RecordRef, ThingInput,
};
use polymodel_api::space_polymodel::library::stage_file::{StageFileOutput, StageFileOutputStatus};
use polymodel_api::space_polymodel::library::update_thing::{
    ModelInput as UpdateModelInput, PartInput as UpdatePartInput, ThingInput as UpdateThingInput,
    UpdateThingOutput, UpdateThingRequest,
};
use polymodel_api::space_polymodel::library::{Chunk, File};
use polymodel_api::space_polymodel::library::{model::Model, part::Part, thing::Thing};
use serde::Serialize;
use sha2::{Digest, Sha256};
use std::collections::BTreeSet;

use super::error::{AppError, AppResult, db, internal, invalid_request, unauthorized};
use super::state::AppState;
use super::views;
use crate::indexing::projection::{ProjectionEvent, project_eager_record};
use crate::oauth::SqliteAuthStore;

const PROFILE_COLLECTION: &str = "space.polymodel.actor.profile";
const THING_COLLECTION: &str = "space.polymodel.library.thing";
const MODEL_COLLECTION: &str = "space.polymodel.library.model";
const PART_COLLECTION: &str = "space.polymodel.library.part";
const LIKE_COLLECTION: &str = "space.polymodel.graph.like";
const SAVE_COLLECTION: &str = "space.polymodel.graph.save";
const TAG_COLLECTION: &str = "space.polymodel.graph.tag";
const CHUNK_SIZE: usize = 10 * 1024 * 1024;
const MAX_STAGE_FILE_SIZE: usize = 100 * 1024 * 1024;

type UserAgent = Agent<OAuthSession<PublicResolver, SqliteAuthStore>>;

pub(super) async fn bootstrap_profile(
    State(state): State<AppState>,
    ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>,
) -> AppResult<XrpcResponse<BootstrapProfileRequest>> {
    let agent = Agent::from(session);
    let actor = authenticated_did(&agent).await?;
    let outcome = ensure_polymodel_profile(&state, &agent, &actor, true).await?;
    Ok(XrpcResponse(BootstrapProfileOutput {
        cid: outcome.cid,
        profile: outcome.profile,
        status: outcome.status,
        uri: outcome.uri,
        extra_data: None,
    }))
}

pub(super) async fn stage_file(
    State(state): State<AppState>,
    ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>,
    headers: HeaderMap,
    body: Body,
) -> AppResult<Json<StageFileOutput>> {
    let agent = Agent::from(session);
    let actor = authenticated_did(&agent).await?;
    let _ = ensure_polymodel_profile(&state, &agent, &actor, false).await?;

    let mime_type = content_type(&headers);
    let filename = filename_header(&headers);
    let staged = upload_streamed_file(&agent, body, mime_type.as_str()).await?;

    let file = File {
        chunks: staged.chunks,
        digest: Some(SmolStr::from(staged.digest.clone())),
        mime_type: SmolStr::from(mime_type.clone()),
        size: staged.size as i64,
        extra_data: None,
    };
    let upload_id = format!("{}-{}", &staged.digest[..16], staged.size);
    persist_upload(
        &state,
        &actor,
        UploadPersistence {
            upload_id: &upload_id,
            digest: &staged.digest,
            mime_type: &mime_type,
            size: staged.size,
            filename: filename.as_deref(),
            file: &file,
        },
    )
    .await?;

    Ok(Json(StageFileOutput {
        file,
        filename: filename.map(SmolStr::from),
        status: Some(StageFileOutputStatus::Uploaded),
        upload_id: SmolStr::from(upload_id),
        extra_data: None,
    }))
}

pub(super) async fn publish_thing(
    State(state): State<AppState>,
    ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>,
    ExtractXrpc(req): ExtractXrpc<PublishThingRequest>,
) -> AppResult<XrpcResponse<PublishThingRequest>> {
    let agent = Agent::from(session);
    let actor = authenticated_did(&agent).await?;
    let _ = ensure_polymodel_profile(&state, &agent, &actor, false).await?;
    let _write_guard = state.write_lock.lock().await;
    let out = publish_composition(&state, &agent, &actor, req.thing).await?;
    Ok(XrpcResponse(out))
}

pub(super) async fn update_thing(
    State(state): State<AppState>,
    ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>,
    ExtractXrpc(req): ExtractXrpc<UpdateThingRequest>,
) -> AppResult<XrpcResponse<UpdateThingRequest>> {
    let agent = Agent::from(session);
    let actor = authenticated_did(&agent).await?;
    let _ = ensure_polymodel_profile(&state, &agent, &actor, false).await?;
    let _write_guard = state.write_lock.lock().await;
    validate_owned_uri(&actor, &req.uri, THING_COLLECTION)?;
    let previous = load_owned_composition(&state, &actor, &req.uri).await?;
    let out = update_composition(&state, &agent, &actor, req.uri, req.thing, &previous).await?;
    Ok(XrpcResponse(UpdateThingOutput {
        models: out.models,
        thing: out.thing,
        extra_data: None,
    }))
}

pub(super) async fn delete_thing(
    State(state): State<AppState>,
    ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>,
    ExtractXrpc(req): ExtractXrpc<DeleteThingRequest>,
) -> AppResult<XrpcResponse<DeleteThingRequest>> {
    let agent = Agent::from(session);
    let actor = authenticated_did(&agent).await?;
    let _write_guard = state.write_lock.lock().await;
    validate_owned_uri(&actor, &req.uri, THING_COLLECTION)?;
    let existing = load_owned_composition(&state, &actor, &req.uri).await?;
    let mut deleted_uris = delete_composition_children(&state, &agent, &actor, &existing).await?;
    agent_delete_record::<Thing>(&agent, existing.thing_rkey.clone()).await?;
    eager_delete(
        &state,
        &actor,
        THING_COLLECTION,
        existing.thing_rkey.as_ref(),
    )
    .await?;
    deleted_uris.insert(0, existing.thing_uri);
    Ok(XrpcResponse(DeleteThingOutput {
        deleted: true,
        deleted_uris: Some(deleted_uris),
        extra_data: None,
    }))
}

pub(super) async fn create_like(
    State(state): State<AppState>,
    ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>,
    ExtractXrpc(req): ExtractXrpc<CreateLikeRequest>,
) -> AppResult<XrpcResponse<CreateLikeRequest>> {
    let agent = Agent::from(session);
    let actor = authenticated_did(&agent).await?;
    let _ = ensure_polymodel_profile(&state, &agent, &actor, false).await?;
    validate_subject(&state, &req.subject.uri).await?;
    if let Some(existing) =
        lookup_existing_social(&state, &actor, LIKE_COLLECTION, req.subject.uri.as_ref()).await?
    {
        return Ok(XrpcResponse(CreateLikeOutput {
            cid: existing.cid,
            created: Some(false),
            uri: existing.uri,
            extra_data: None,
        }));
    }
    let record = Like {
        created_at: Datetime::now(),
        subject: req.subject,
        extra_data: None,
    };
    let out = agent
        .create_record(record.clone(), None)
        .await
        .map_err(|e| agent_error("create like", e))?;
    let rkey = rkey_from_uri(&out.uri)?;
    eager_project_record(
        &state,
        &actor,
        LIKE_COLLECTION,
        rkey.as_ref(),
        Some(out.cid.clone()),
        &record,
        "create",
    )
    .await?;
    Ok(XrpcResponse(CreateLikeOutput {
        cid: out.cid,
        created: Some(true),
        uri: out.uri,
        extra_data: None,
    }))
}

pub(super) async fn delete_like(
    State(state): State<AppState>,
    ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>,
    ExtractXrpc(req): ExtractXrpc<DeleteLikeRequest>,
) -> AppResult<XrpcResponse<DeleteLikeRequest>> {
    let agent = Agent::from(session);
    let actor = authenticated_did(&agent).await?;
    let uri = match (req.uri, req.subject) {
        (Some(uri), _) => uri,
        (None, Some(subject)) => {
            lookup_social_uri(&state, &actor, LIKE_COLLECTION, subject.uri.as_ref()).await?
        }
        (None, None) => return Err(invalid_request("deleteLike requires uri or subject")),
    };
    delete_social::<Like>(&state, &agent, &actor, LIKE_COLLECTION, &uri).await?;
    Ok(XrpcResponse(DeleteLikeOutput {
        deleted: true,
        uri: Some(uri),
        extra_data: None,
    }))
}

pub(super) async fn create_save(
    State(state): State<AppState>,
    ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>,
    ExtractXrpc(req): ExtractXrpc<CreateSaveRequest>,
) -> AppResult<XrpcResponse<CreateSaveRequest>> {
    let agent = Agent::from(session);
    let actor = authenticated_did(&agent).await?;
    let _ = ensure_polymodel_profile(&state, &agent, &actor, false).await?;
    validate_subject(&state, &req.subject.uri).await?;
    if let Some(existing) =
        lookup_existing_social(&state, &actor, SAVE_COLLECTION, req.subject.uri.as_ref()).await?
    {
        return Ok(XrpcResponse(CreateSaveOutput {
            cid: existing.cid,
            created: Some(false),
            uri: existing.uri,
            extra_data: None,
        }));
    }
    let record = Save {
        created_at: Datetime::now(),
        note: req.note,
        subject: req.subject,
        extra_data: None,
    };
    let out = agent
        .create_record(record.clone(), None)
        .await
        .map_err(|e| agent_error("create save", e))?;
    let rkey = rkey_from_uri(&out.uri)?;
    eager_project_record(
        &state,
        &actor,
        SAVE_COLLECTION,
        rkey.as_ref(),
        Some(out.cid.clone()),
        &record,
        "create",
    )
    .await?;
    Ok(XrpcResponse(CreateSaveOutput {
        cid: out.cid,
        created: Some(true),
        uri: out.uri,
        extra_data: None,
    }))
}

pub(super) async fn delete_save(
    State(state): State<AppState>,
    ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>,
    ExtractXrpc(req): ExtractXrpc<DeleteSaveRequest>,
) -> AppResult<XrpcResponse<DeleteSaveRequest>> {
    let agent = Agent::from(session);
    let actor = authenticated_did(&agent).await?;
    let uri = match (req.uri, req.subject) {
        (Some(uri), _) => uri,
        (None, Some(subject)) => {
            lookup_social_uri(&state, &actor, SAVE_COLLECTION, subject.uri.as_ref()).await?
        }
        (None, None) => return Err(invalid_request("deleteSave requires uri or subject")),
    };
    delete_social::<Save>(&state, &agent, &actor, SAVE_COLLECTION, &uri).await?;
    Ok(XrpcResponse(DeleteSaveOutput {
        deleted: true,
        uri: Some(uri),
        extra_data: None,
    }))
}

pub(super) async fn create_tag(
    State(state): State<AppState>,
    ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>,
    ExtractXrpc(req): ExtractXrpc<CreateTagRequest>,
) -> AppResult<XrpcResponse<CreateTagRequest>> {
    let agent = Agent::from(session);
    let actor = authenticated_did(&agent).await?;
    let _ = ensure_polymodel_profile(&state, &agent, &actor, false).await?;
    validate_subject(&state, &req.subject.uri).await?;
    let record = Tag {
        created_at: Datetime::now(),
        subject: req.subject,
        tag: req.tag,
        extra_data: None,
    };
    let out = agent
        .create_record(record.clone(), None)
        .await
        .map_err(|e| agent_error("create tag", e))?;
    let rkey = rkey_from_uri(&out.uri)?;
    eager_project_record(
        &state,
        &actor,
        TAG_COLLECTION,
        rkey.as_ref(),
        Some(out.cid.clone()),
        &record,
        "create",
    )
    .await?;
    Ok(XrpcResponse(CreateTagOutput {
        cid: out.cid,
        created: Some(true),
        uri: out.uri,
        extra_data: None,
    }))
}

pub(super) async fn delete_tag(
    State(state): State<AppState>,
    ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>,
    ExtractXrpc(req): ExtractXrpc<DeleteTagRequest>,
) -> AppResult<XrpcResponse<DeleteTagRequest>> {
    let agent = Agent::from(session);
    let actor = authenticated_did(&agent).await?;
    delete_social::<Tag>(&state, &agent, &actor, TAG_COLLECTION, &req.uri).await?;
    Ok(XrpcResponse(DeleteTagOutput {
        deleted: true,
        uri: Some(req.uri),
        extra_data: None,
    }))
}

struct BootstrapOutcome {
    profile: ProfileView,
    status: BootstrapProfileOutputStatus,
    uri: Option<AtUri>,
    cid: Option<Cid>,
}

async fn authenticated_did(agent: &UserAgent) -> AppResult<Did> {
    agent
        .session_info()
        .await
        .map(|(did, _)| did)
        .ok_or_else(|| unauthorized("OAuth session is required"))
}

async fn ensure_polymodel_profile(
    state: &AppState,
    agent: &UserAgent,
    actor: &Did,
    allow_create: bool,
) -> AppResult<BootstrapOutcome> {
    if let Some(profile) = views::polymodel_profile_view(state, actor).await? {
        return Ok(BootstrapOutcome {
            profile,
            status: BootstrapProfileOutputStatus::ExistingLocal,
            uri: None,
            cid: None,
        });
    }

    let profile_uri = at_uri_owned(actor, PROFILE_COLLECTION, "self")?;
    match agent.get_record::<PolymodelProfile, _>(&profile_uri).await {
        Ok(resp) => {
            let output = resp
                .into_output()
                .map_err(|e| internal(format!("profile record decode failed: {e}")))?;
            eager_project_record(
                state,
                actor,
                PROFILE_COLLECTION,
                "self",
                output.cid.clone(),
                &output.value,
                "create",
            )
            .await?;
            let profile = views::polymodel_profile_view(state, actor)
                .await?
                .ok_or_else(|| internal("projected profile was not readable"))?;
            return Ok(BootstrapOutcome {
                profile,
                status: BootstrapProfileOutputStatus::ExistingRemote,
                uri: Some(output.uri),
                cid: output.cid,
            });
        }
        Err(e) if client_error_is_not_found(&e) => {}
        Err(e) => return Err(client_error("fetch polymodel profile", e)),
    }

    if !allow_create {
        return Err(invalid_request(
            "Polymodel profile has not been bootstrapped",
        ));
    }

    let bsky_uri = at_uri_owned(actor, "app.bsky.actor.profile", "self")?;
    let bsky = match agent.get_record::<BskyProfile, _>(&bsky_uri).await {
        Ok(resp) => {
            resp.into_output()
                .map_err(|e| internal(format!("bsky profile decode failed: {e}")))?
                .value
        }
        Err(e) if client_error_is_not_found(&e) => BskyProfile {
            avatar: None,
            banner: None,
            created_at: None,
            description: None,
            display_name: None,
            joined_via_starter_pack: None,
            labels: None,
            pinned_post: None,
            pronouns: None,
            website: None,
            extra_data: None,
        },
        Err(e) => return Err(client_error("fetch bsky profile", e)),
    };

    let profile = PolymodelProfile {
        avatar: bsky.avatar,
        default_license: None,
        description: bsky.description,
        display_name: bsky.display_name,
        links: None,
        printers: None,
        pronouns: bsky.pronouns,
        extra_data: None,
    };
    let create = agent
        .create_record(profile.clone(), Some(record_key("self")?))
        .await;
    let output = match create {
        Ok(out) => {
            eager_project_record(
                state,
                actor,
                PROFILE_COLLECTION,
                "self",
                Some(out.cid.clone()),
                &profile,
                "create",
            )
            .await?;
            BootstrapOutcome {
                profile: views::polymodel_profile_view(state, actor)
                    .await?
                    .ok_or_else(|| internal("created profile was not readable"))?,
                status: BootstrapProfileOutputStatus::Created,
                uri: Some(out.uri),
                cid: Some(out.cid),
            }
        }
        Err(e) if agent_error_is_conflict(&e) || agent_error_is_not_found(&e) => {
            let resp = agent
                .get_record::<PolymodelProfile, _>(&profile_uri)
                .await
                .map_err(|e| client_error("refetch raced profile", e))?;
            let remote = resp
                .into_output()
                .map_err(|e| internal(format!("raced profile decode failed: {e}")))?;
            eager_project_record(
                state,
                actor,
                PROFILE_COLLECTION,
                "self",
                remote.cid.clone(),
                &remote.value,
                "create",
            )
            .await?;
            BootstrapOutcome {
                profile: views::polymodel_profile_view(state, actor)
                    .await?
                    .ok_or_else(|| internal("raced profile was not readable"))?,
                status: BootstrapProfileOutputStatus::ExistingRemote,
                uri: Some(remote.uri),
                cid: remote.cid,
            }
        }
        Err(e) => return Err(agent_error("create polymodel profile", e)),
    };
    Ok(output)
}

async fn publish_composition(
    state: &AppState,
    agent: &UserAgent,
    actor: &Did,
    thing: ThingInput,
) -> AppResult<PublishThingOutput> {
    let mut model_results = Vec::new();
    let mut model_refs = Vec::new();

    for model_input in thing.models {
        let (model_ref, result) = publish_model(state, agent, actor, model_input).await?;
        model_refs.push(model_ref);
        model_results.push(result);
    }

    let record = Thing {
        cover: thing.cover,
        created_at: Datetime::now(),
        derived_from: thing.derived_from,
        instructions: thing.instructions,
        intended_to_work_with: thing.intended_to_work_with,
        license: thing.license,
        models: Some(model_refs),
        name: thing.name,
        previews: thing.previews,
        summary: thing.summary,
        tags: thing.tags,
        extra_data: None,
    };

    let out = agent
        .create_record(record.clone(), None)
        .await
        .map_err(|e| agent_error("create thing", e))?;
    let rkey = rkey_from_uri(&out.uri)?;
    eager_project_record(
        state,
        actor,
        THING_COLLECTION,
        rkey.as_ref(),
        Some(out.cid.clone()),
        &record,
        "create",
    )
    .await?;
    let created = RecordRef {
        cid: out.cid,
        uri: out.uri,
        extra_data: None,
    };

    Ok(PublishThingOutput {
        thing: created,
        models: model_results,
        extra_data: None,
    })
}

async fn update_composition(
    state: &AppState,
    agent: &UserAgent,
    actor: &Did,
    thing_uri: AtUri,
    thing: UpdateThingInput,
    previous: &ExistingComposition,
) -> AppResult<PublishThingOutput> {
    let mut model_results = Vec::new();
    let mut model_refs = Vec::new();

    for model_input in thing.models {
        let (model_ref, result) = update_model(state, agent, actor, model_input, previous).await?;
        model_refs.push(model_ref);
        model_results.push(result);
    }

    let record = Thing {
        cover: thing.cover,
        created_at: Datetime::now(),
        derived_from: thing.derived_from,
        instructions: thing.instructions,
        intended_to_work_with: thing.intended_to_work_with,
        license: thing.license,
        models: Some(model_refs),
        name: thing.name,
        previews: thing.previews,
        summary: thing.summary,
        tags: thing.tags,
        extra_data: None,
    };

    let rkey = rkey_from_uri(&thing_uri)?;
    let out = agent
        .put_record::<Thing>(rkey.clone(), record.clone())
        .await
        .map_err(|e| agent_error("put thing", e))?;
    eager_project_record(
        state,
        actor,
        THING_COLLECTION,
        rkey.as_ref(),
        Some(out.cid.clone()),
        &record,
        "update",
    )
    .await?;

    Ok(PublishThingOutput {
        thing: RecordRef {
            cid: out.cid,
            uri: out.uri,
            extra_data: None,
        },
        models: model_results,
        extra_data: None,
    })
}

async fn update_model(
    state: &AppState,
    agent: &UserAgent,
    actor: &Did,
    model_input: UpdateModelInput,
    previous: &ExistingComposition,
) -> AppResult<(StrongRef, ModelPublishResult)> {
    let mut part_refs = Vec::new();
    let mut part_record_refs = Vec::new();
    for part_input in model_input.parts {
        let part_ref = update_part(state, agent, actor, part_input, previous).await?;
        part_record_refs.push(RecordRef {
            uri: part_ref.uri.clone(),
            cid: part_ref.cid.clone(),
            extra_data: None,
        });
        part_refs.push(part_ref);
    }

    let model = Model {
        cover: model_input.cover,
        created_at: Datetime::now(),
        derived_from: model_input.derived_from,
        instructions: model_input.instructions,
        intended_to_work_with: model_input.intended_to_work_with,
        license: model_input.license,
        name: model_input.name,
        parts: part_refs,
        previews: model_input.previews,
        summary: model_input.summary,
        tags: model_input.tags,
        extra_data: None,
    };

    let (uri, cid, action) = if let Some(uri) = model_input.uri {
        validate_owned_uri(actor, &uri, MODEL_COLLECTION)?;
        let rkey = rkey_from_uri(&uri)?;
        validate_existing_update_child(
            state,
            actor,
            &uri,
            &rkey,
            MODEL_COLLECTION,
            &previous.models,
        )
        .await?;
        let out = agent
            .put_record::<Model>(rkey.clone(), model.clone())
            .await
            .map_err(|e| agent_error("put model", e))?;
        eager_project_record(
            state,
            actor,
            MODEL_COLLECTION,
            rkey.as_ref(),
            Some(out.cid.clone()),
            &model,
            "update",
        )
        .await?;
        (out.uri, out.cid, "update")
    } else {
        let out = agent
            .create_record(model.clone(), None)
            .await
            .map_err(|e| agent_error("create model", e))?;
        let rkey = rkey_from_uri(&out.uri)?;
        eager_project_record(
            state,
            actor,
            MODEL_COLLECTION,
            rkey.as_ref(),
            Some(out.cid.clone()),
            &model,
            "create",
        )
        .await?;
        (out.uri, out.cid, "create")
    };
    tracing::debug!(action, uri = %uri, "updated thing model composition member");
    let model_ref = strong_ref(uri.clone(), cid.clone());
    Ok((
        model_ref,
        ModelPublishResult {
            model: RecordRef {
                uri,
                cid,
                extra_data: None,
            },
            parts: part_record_refs,
            extra_data: None,
        },
    ))
}

async fn update_part(
    state: &AppState,
    agent: &UserAgent,
    actor: &Did,
    input: UpdatePartInput,
    previous: &ExistingComposition,
) -> AppResult<StrongRef> {
    let file = match (input.file, input.upload_id) {
        (Some(file), _) => file,
        (None, Some(upload_id)) => load_upload_file(state, actor, upload_id.as_ref()).await?,
        (None, None) => return Err(invalid_request("part requires file or uploadId")),
    };
    let part = Part {
        created_at: Datetime::now(),
        derived_from: input.derived_from,
        dimensions: input.dimensions,
        file,
        format: input.format,
        intended_to_work_with: input.intended_to_work_with,
        name: input.name,
        notes: input.notes,
        previews: input.previews,
        print_settings: input.print_settings,
        units: input.units,
        extra_data: None,
    };

    if let Some(uri) = input.uri {
        validate_owned_uri(actor, &uri, PART_COLLECTION)?;
        let rkey = rkey_from_uri(&uri)?;
        validate_existing_update_child(state, actor, &uri, &rkey, PART_COLLECTION, &previous.parts)
            .await?;
        let out = agent
            .put_record::<Part>(rkey.clone(), part.clone())
            .await
            .map_err(|e| agent_error("put part", e))?;
        eager_project_record(
            state,
            actor,
            PART_COLLECTION,
            rkey.as_ref(),
            Some(out.cid.clone()),
            &part,
            "update",
        )
        .await?;
        return Ok(strong_ref(out.uri, out.cid));
    }

    let out = agent
        .create_record(part.clone(), None)
        .await
        .map_err(|e| agent_error("create part", e))?;
    let rkey = rkey_from_uri(&out.uri)?;
    eager_project_record(
        state,
        actor,
        PART_COLLECTION,
        rkey.as_ref(),
        Some(out.cid.clone()),
        &part,
        "create",
    )
    .await?;
    Ok(strong_ref(out.uri, out.cid))
}

async fn publish_model(
    state: &AppState,
    agent: &UserAgent,
    actor: &Did,
    model_input: ModelInput,
) -> AppResult<(StrongRef, ModelPublishResult)> {
    let mut part_refs = Vec::new();
    let mut part_record_refs = Vec::new();
    for part_input in model_input.parts {
        let part_ref = publish_part(state, agent, actor, part_input).await?;
        part_record_refs.push(RecordRef {
            uri: part_ref.uri.clone(),
            cid: part_ref.cid.clone(),
            extra_data: None,
        });
        part_refs.push(part_ref);
    }
    let model = Model {
        cover: model_input.cover,
        created_at: Datetime::now(),
        derived_from: model_input.derived_from,
        instructions: model_input.instructions,
        intended_to_work_with: model_input.intended_to_work_with,
        license: model_input.license,
        name: model_input.name,
        parts: part_refs,
        previews: model_input.previews,
        summary: model_input.summary,
        tags: model_input.tags,
        extra_data: None,
    };
    let out = agent
        .create_record(model.clone(), None)
        .await
        .map_err(|e| agent_error("create model", e))?;
    let rkey = rkey_from_uri(&out.uri)?;
    eager_project_record(
        state,
        actor,
        MODEL_COLLECTION,
        rkey.as_ref(),
        Some(out.cid.clone()),
        &model,
        "create",
    )
    .await?;
    let model_ref = strong_ref(out.uri.clone(), out.cid.clone());
    let record_ref = RecordRef {
        uri: out.uri,
        cid: out.cid,
        extra_data: None,
    };
    Ok((
        model_ref,
        ModelPublishResult {
            model: record_ref,
            parts: part_record_refs,
            extra_data: None,
        },
    ))
}

async fn publish_part(
    state: &AppState,
    agent: &UserAgent,
    actor: &Did,
    input: polymodel_api::space_polymodel::library::publish_thing::PartInput,
) -> AppResult<StrongRef> {
    let file = match (input.file, input.upload_id) {
        (Some(file), _) => file,
        (None, Some(upload_id)) => load_upload_file(state, actor, upload_id.as_ref()).await?,
        (None, None) => return Err(invalid_request("part requires file or uploadId")),
    };
    let part = Part {
        created_at: Datetime::now(),
        derived_from: input.derived_from,
        dimensions: input.dimensions,
        file,
        format: input.format,
        intended_to_work_with: input.intended_to_work_with,
        name: input.name,
        notes: input.notes,
        previews: input.previews,
        print_settings: input.print_settings,
        units: input.units,
        extra_data: None,
    };
    let out = agent
        .create_record(part.clone(), None)
        .await
        .map_err(|e| agent_error("create part", e))?;
    let rkey = rkey_from_uri(&out.uri)?;
    eager_project_record(
        state,
        actor,
        PART_COLLECTION,
        rkey.as_ref(),
        Some(out.cid.clone()),
        &part,
        "create",
    )
    .await?;
    Ok(strong_ref(out.uri, out.cid))
}

async fn delete_social<R>(
    state: &AppState,
    agent: &UserAgent,
    actor: &Did,
    collection: &str,
    uri: &AtUri,
) -> AppResult<()>
where
    R: Collection + Serialize,
{
    validate_owned_uri(actor, uri, collection)?;
    let rkey = rkey_from_uri(uri)?;
    agent_delete_record::<R>(agent, rkey.clone()).await?;
    eager_delete(state, actor, collection, rkey.as_ref()).await
}

struct ExistingComposition {
    thing_uri: AtUri,
    thing_rkey: RecordKey<Rkey>,
    models: Vec<CompositionChild>,
    parts: Vec<CompositionChild>,
}

struct CompositionChild {
    rkey: RecordKey<Rkey>,
    uri: AtUri,
}

#[derive(sqlx::FromRow)]
struct CompositionRecordRow {
    did: String,
    rkey: String,
    uri: String,
}

async fn load_owned_composition(
    state: &AppState,
    actor: &Did,
    thing_uri: &AtUri,
) -> AppResult<ExistingComposition> {
    let thing = db(sqlx::query_as!(
        CompositionRecordRow,
        "SELECT did, rkey, uri FROM things WHERE uri = ?",
        thing_uri.as_ref()
    )
    .fetch_optional(&state.pool)
    .await)?
    .ok_or_else(|| invalid_request("thing is not indexed locally"))?;
    ensure_row_owned(actor, &thing, THING_COLLECTION)?;

    let models = db(sqlx::query_as!(
        CompositionRecordRow,
        "SELECT m.did, m.rkey, m.uri FROM thing_models tm \
         JOIN models m ON m.uri = tm.model_uri \
         WHERE tm.thing_uri = ? ORDER BY tm.position",
        thing_uri.as_ref()
    )
    .fetch_all(&state.pool)
    .await)?;
    let parts = db(sqlx::query_as!(
        CompositionRecordRow,
        "SELECT p.did, p.rkey, p.uri FROM thing_models tm \
         JOIN model_parts mp ON mp.model_uri = tm.model_uri \
         JOIN parts p ON p.uri = mp.part_uri \
         WHERE tm.thing_uri = ? ORDER BY tm.position, mp.position",
        thing_uri.as_ref()
    )
    .fetch_all(&state.pool)
    .await)?;

    let thing_uri = at_uri_from_row(&thing)?;
    let mut model_children = Vec::new();
    let mut seen_models = BTreeSet::new();
    for row in models {
        ensure_row_owned(actor, &row, MODEL_COLLECTION)?;
        if seen_models.insert(row.uri.clone()) {
            model_children.push(CompositionChild {
                rkey: record_key(&row.rkey)?,
                uri: at_uri_from_row(&row)?,
            });
        }
    }
    let mut part_children = Vec::new();
    let mut seen_parts = BTreeSet::new();
    for row in parts {
        ensure_row_owned(actor, &row, PART_COLLECTION)?;
        if seen_parts.insert(row.uri.clone()) {
            part_children.push(CompositionChild {
                rkey: record_key(&row.rkey)?,
                uri: at_uri_from_row(&row)?,
            });
        }
    }

    Ok(ExistingComposition {
        thing_uri,
        thing_rkey: record_key(&thing.rkey)?,
        models: model_children,
        parts: part_children,
    })
}

async fn delete_composition_children(
    state: &AppState,
    agent: &UserAgent,
    actor: &Did,
    composition: &ExistingComposition,
) -> AppResult<Vec<AtUri>> {
    let mut deleted_uris = Vec::new();
    for model in &composition.models {
        if model_reference_count(state, &model.uri).await? > 1 {
            continue;
        }
        agent_delete_record::<Model>(agent, model.rkey.clone()).await?;
        eager_delete(state, actor, MODEL_COLLECTION, model.rkey.as_ref()).await?;
        deleted_uris.push(model.uri.clone());
    }
    for part in &composition.parts {
        if part_reference_count(state, &part.uri).await? > 0 {
            continue;
        }
        agent_delete_record::<Part>(agent, part.rkey.clone()).await?;
        eager_delete(state, actor, PART_COLLECTION, part.rkey.as_ref()).await?;
        deleted_uris.push(part.uri.clone());
    }
    Ok(deleted_uris)
}

async fn model_reference_count(state: &AppState, uri: &AtUri) -> AppResult<i64> {
    db(sqlx::query_scalar!(
        "SELECT COUNT(*) AS count FROM thing_models WHERE model_uri = ?",
        uri.as_ref()
    )
    .fetch_one(&state.pool)
    .await)
}

async fn part_reference_count(state: &AppState, uri: &AtUri) -> AppResult<i64> {
    db(sqlx::query_scalar!(
        "SELECT COUNT(*) AS count FROM model_parts WHERE part_uri = ?",
        uri.as_ref()
    )
    .fetch_one(&state.pool)
    .await)
}

fn ensure_row_owned(actor: &Did, row: &impl OwnedRow, collection: &str) -> AppResult<()> {
    if row.did() != actor.as_ref() {
        return Err(invalid_request(format!(
            "{collection} composition record is not owned by the authenticated actor"
        )));
    }
    Ok(())
}

trait OwnedRow {
    fn did(&self) -> &str;
    fn uri(&self) -> &str;
}

impl OwnedRow for CompositionRecordRow {
    fn did(&self) -> &str {
        &self.did
    }
    fn uri(&self) -> &str {
        &self.uri
    }
}

fn at_uri_from_row(row: &impl OwnedRow) -> AppResult<AtUri> {
    AtUri::new_owned(row.uri()).map_err(|e| internal(format!("stored at-uri is invalid: {e}")))
}

fn strong_ref(uri: AtUri, cid: Cid) -> StrongRef {
    StrongRef {
        uri,
        cid,
        extra_data: None,
    }
}

fn record_key(value: &str) -> AppResult<RecordKey<Rkey>> {
    RecordKey::any(SmolStr::from(value))
        .map_err(|e| invalid_request(format!("invalid rkey {value:?}: {e}")))
}

fn rkey_from_uri(uri: &AtUri) -> AppResult<RecordKey<Rkey>> {
    let rkey = uri
        .rkey()
        .ok_or_else(|| invalid_request("at-uri is missing record key"))?;
    record_key(rkey.as_ref())
}

fn at_uri_owned(actor: &Did, collection: &str, rkey: &str) -> AppResult<AtUri> {
    AtUri::new_owned(format!("at://{}/{}/{}", actor.as_ref(), collection, rkey))
        .map_err(|e| invalid_request(format!("invalid at-uri: {e}")))
}

fn validate_owned_uri(actor: &Did, uri: &AtUri, expected_collection: &str) -> AppResult<()> {
    match uri.authority() {
        AtIdentifier::Did(did) if did.as_ref() == actor.as_ref() => {}
        _ => {
            return Err(invalid_request(
                "record URI is not owned by the authenticated actor",
            ));
        }
    }
    let collection = uri
        .collection()
        .ok_or_else(|| invalid_request("record URI missing collection"))?;
    if collection.as_ref() != expected_collection {
        return Err(invalid_request(format!(
            "expected {expected_collection} URI"
        )));
    }
    if uri.rkey().is_none() {
        return Err(invalid_request("record URI missing rkey"));
    }
    Ok(())
}

async fn validate_existing_update_child(
    state: &AppState,
    actor: &Did,
    uri: &AtUri,
    rkey: &RecordKey<Rkey>,
    collection: &str,
    current_children: &[CompositionChild],
) -> AppResult<()> {
    if current_children.iter().any(|child| &child.rkey == rkey) {
        return Ok(());
    }
    let exists = match collection {
        MODEL_COLLECTION => db(sqlx::query_scalar!(
            "SELECT EXISTS(SELECT 1 FROM models WHERE did = ? AND uri = ?)",
            actor.as_ref(),
            uri.as_ref()
        )
        .fetch_one(&state.pool)
        .await)?,
        PART_COLLECTION => db(sqlx::query_scalar!(
            "SELECT EXISTS(SELECT 1 FROM parts WHERE did = ? AND uri = ?)",
            actor.as_ref(),
            uri.as_ref()
        )
        .fetch_one(&state.pool)
        .await)?,
        _ => return Err(invalid_request("unsupported composition child collection")),
    };
    if exists == 0 {
        return Err(invalid_request(
            "existing child uri is not indexed as an owned composition record",
        ));
    }
    Ok(())
}

async fn validate_subject(state: &AppState, uri: &AtUri) -> AppResult<()> {
    let collection = uri
        .collection()
        .ok_or_else(|| invalid_request("subject URI missing collection"))?;
    let table = match collection.as_ref() {
        THING_COLLECTION => "things",
        MODEL_COLLECTION => "models",
        PART_COLLECTION => "parts",
        _ => {
            return Err(invalid_request(
                "subject must be a Polymodel thing, model, or part",
            ));
        }
    };
    let exists = match table {
        "things" => db(sqlx::query_scalar!(
            "SELECT EXISTS(SELECT 1 FROM things WHERE uri = ?)",
            uri.as_ref()
        )
        .fetch_one(&state.pool)
        .await)?,
        "models" => db(sqlx::query_scalar!(
            "SELECT EXISTS(SELECT 1 FROM models WHERE uri = ?)",
            uri.as_ref()
        )
        .fetch_one(&state.pool)
        .await)?,
        "parts" => db(sqlx::query_scalar!(
            "SELECT EXISTS(SELECT 1 FROM parts WHERE uri = ?)",
            uri.as_ref()
        )
        .fetch_one(&state.pool)
        .await)?,
        _ => unreachable!(),
    };
    if exists == 0 {
        return Err(invalid_request("subject is not indexed locally"));
    }
    Ok(())
}

async fn eager_project_record<R: Serialize>(
    state: &AppState,
    actor: &Did,
    collection: &str,
    rkey: &str,
    cid: Option<Cid>,
    record: &R,
    action: &str,
) -> AppResult<()> {
    let record =
        serde_json::to_value(record).map_err(|e| internal(format!("record serialize: {e}")))?;
    let event = ProjectionEvent {
        seq: 0,
        did: actor.as_ref().to_string(),
        collection: collection.to_string(),
        rkey: rkey.to_string(),
        action: action.to_string(),
        record: Some(record),
        cid: cid.map(|c| c.to_string()),
    };
    project_eager_record(&state.pool, &event)
        .await
        .map_err(|e| internal(format!("eager projection failed: {e}")))
}

async fn eager_delete(
    state: &AppState,
    actor: &Did,
    collection: &str,
    rkey: &str,
) -> AppResult<()> {
    let event = ProjectionEvent {
        seq: 0,
        did: actor.as_ref().to_string(),
        collection: collection.to_string(),
        rkey: rkey.to_string(),
        action: "delete".to_string(),
        record: None,
        cid: None,
    };
    project_eager_record(&state.pool, &event)
        .await
        .map_err(|e| internal(format!("eager delete projection failed: {e}")))
}

struct StreamedUpload {
    chunks: Vec<Chunk>,
    digest: String,
    size: usize,
}

async fn upload_streamed_file(
    agent: &UserAgent,
    mut body: Body,
    mime_type: &str,
) -> AppResult<StreamedUpload> {
    let mut hasher = Sha256::new();
    let mut chunks = Vec::new();
    let mut pending = Vec::with_capacity(CHUNK_SIZE);
    let mut size = 0usize;

    while let Some(frame) = body
        .frame()
        .await
        .transpose()
        .map_err(|e| invalid_request(format!("stageFile body read failed: {e}")))?
    {
        let Ok(data) = frame.into_data() else {
            continue;
        };
        size = size
            .checked_add(data.len())
            .ok_or_else(|| invalid_request("stageFile body is too large"))?;
        if size > MAX_STAGE_FILE_SIZE {
            return Err(invalid_request(format!(
                "stageFile body exceeds the {} byte limit",
                MAX_STAGE_FILE_SIZE
            )));
        }
        hasher.update(&data);
        let mut remaining = data.as_ref();
        while !remaining.is_empty() {
            let take = (CHUNK_SIZE - pending.len()).min(remaining.len());
            pending.extend_from_slice(&remaining[..take]);
            remaining = &remaining[take..];
            if pending.len() == CHUNK_SIZE {
                let offset = size - remaining.len() - pending.len();
                upload_one_chunk(
                    agent,
                    mime_type,
                    &mut chunks,
                    offset,
                    std::mem::take(&mut pending),
                )
                .await?;
                pending = Vec::with_capacity(CHUNK_SIZE);
            }
        }
    }

    if size == 0 {
        return Err(invalid_request(
            "stageFile requires a non-empty raw byte body",
        ));
    }
    if !pending.is_empty() {
        let offset = size - pending.len();
        upload_one_chunk(agent, mime_type, &mut chunks, offset, pending).await?;
    }
    let digest = hasher
        .finalize()
        .iter()
        .map(|b| format!("{b:02x}"))
        .collect();
    Ok(StreamedUpload {
        chunks,
        digest,
        size,
    })
}

async fn upload_one_chunk(
    agent: &UserAgent,
    mime_type: &str,
    chunks: &mut Vec<Chunk>,
    offset: usize,
    bytes: Vec<u8>,
) -> AppResult<()> {
    let size = bytes.len();
    let blob = agent_upload_blob(agent, bytes, mime_type).await?;
    chunks.push(Chunk {
        blob: BlobRef::from(blob),
        offset: offset as i64,
        size: size as i64,
        extra_data: None,
    });
    Ok(())
}

async fn agent_upload_blob(
    agent: &UserAgent,
    bytes: Vec<u8>,
    mime_type: &str,
) -> AppResult<jacquard_common::types::blob::Blob> {
    agent
        .upload_blob(bytes, MimeType::new(mime_type))
        .await
        .map_err(|e| agent_error("upload blob", e))
}

async fn agent_delete_record<R>(agent: &UserAgent, rkey: RecordKey<Rkey>) -> AppResult<()>
where
    R: Collection + Serialize,
{
    agent
        .delete_record::<R>(rkey)
        .await
        .map(|_| ())
        .map_err(|e| agent_error("delete record", e))
}

struct UploadPersistence<'a> {
    upload_id: &'a str,
    digest: &'a str,
    mime_type: &'a str,
    size: usize,
    filename: Option<&'a str>,
    file: &'a File,
}

async fn persist_upload(
    state: &AppState,
    actor: &Did,
    upload: UploadPersistence<'_>,
) -> AppResult<()> {
    let file_json = serde_json::to_string(upload.file)
        .map_err(|e| internal(format!("file metadata serialize: {e}")))?;
    let chunks_json = serde_json::to_string(&upload.file.chunks)
        .map_err(|e| internal(format!("chunk metadata serialize: {e}")))?;
    let now = chrono::Utc::now().timestamp_nanos_opt().unwrap_or_default();
    db(sqlx::query!(
        "INSERT INTO upload_staging (owner_did, upload_id, sha256, mime_type, size, filename, status, file_json, chunks_json, created_at, updated_at) \
         VALUES (?, ?, ?, ?, ?, ?, 'uploaded', ?, ?, ?, ?) \
         ON CONFLICT(owner_did, upload_id) DO UPDATE SET sha256 = excluded.sha256, mime_type = excluded.mime_type, size = excluded.size, filename = excluded.filename, status = excluded.status, file_json = excluded.file_json, chunks_json = excluded.chunks_json, updated_at = excluded.updated_at",
        actor.as_ref(),
        upload.upload_id,
        upload.digest,
        upload.mime_type,
        upload.size as i64,
        upload.filename,
        file_json,
        chunks_json,
        now,
        now,
    ).execute(&state.pool).await)?;
    Ok(())
}

async fn load_upload_file(state: &AppState, actor: &Did, upload_id: &str) -> AppResult<File> {
    #[derive(sqlx::FromRow)]
    struct Row {
        file_json: String,
    }
    let row = db(sqlx::query_as!(Row, "SELECT file_json FROM upload_staging WHERE owner_did = ? AND upload_id = ? AND status = 'uploaded'", actor.as_ref(), upload_id).fetch_optional(&state.pool).await)?
        .ok_or_else(|| invalid_request("unknown or unavailable uploadId"))?;
    serde_json::from_str(&row.file_json)
        .map_err(|e| internal(format!("staged file metadata parse: {e}")))
}

async fn lookup_social_uri(
    state: &AppState,
    actor: &Did,
    collection: &str,
    subject_uri: &str,
) -> AppResult<AtUri> {
    lookup_existing_social(state, actor, collection, subject_uri)
        .await?
        .map(|record| record.uri)
        .ok_or_else(|| invalid_request("social record is not indexed locally"))
}

struct ExistingSocial {
    cid: Cid,
    uri: AtUri,
}

async fn lookup_existing_social(
    state: &AppState,
    actor: &Did,
    collection: &str,
    subject_uri: &str,
) -> AppResult<Option<ExistingSocial>> {
    #[derive(sqlx::FromRow)]
    struct Row {
        cid: String,
        rkey: String,
    }
    let row = match collection {
        LIKE_COLLECTION => db(sqlx::query_as!(
            Row,
            "SELECT cid, rkey FROM likes WHERE did = ? AND subject_uri = ?",
            actor.as_ref(),
            subject_uri
        )
        .fetch_optional(&state.pool)
        .await)?,
        SAVE_COLLECTION => db(sqlx::query_as!(
            Row,
            "SELECT cid, rkey FROM saves WHERE did = ? AND subject_uri = ?",
            actor.as_ref(),
            subject_uri
        )
        .fetch_optional(&state.pool)
        .await)?,
        _ => return Err(invalid_request("unsupported social lookup")),
    };
    let existing = row
        .map(|row| {
            let uri = at_uri_owned(actor, collection, &row.rkey)?;
            let cid = Cid::new_owned(row.cid.as_bytes())
                .map_err(|e| internal(format!("stored social cid is invalid: {e}")))?;
            Ok(ExistingSocial { cid, uri })
        })
        .transpose()?;
    Ok(existing)
}

fn content_type(headers: &HeaderMap) -> String {
    headers
        .get(header::CONTENT_TYPE)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.split(';').next())
        .map(str::trim)
        .filter(|v| !v.is_empty())
        .unwrap_or("application/octet-stream")
        .to_string()
}

fn filename_header(headers: &HeaderMap) -> Option<String> {
    headers
        .get("x-polymodel-filename")
        .and_then(|v| v.to_str().ok())
        .map(str::trim)
        .filter(|v| !v.is_empty())
        .map(ToOwned::to_owned)
}

#[cfg(test)]
fn hex_sha256(bytes: &[u8]) -> String {
    let digest = Sha256::digest(bytes);
    digest.iter().map(|b| format!("{b:02x}")).collect()
}

fn agent_error(operation: &str, err: jacquard::client::AgentError) -> AppError {
    if agent_error_is_auth(&err) {
        tracing::warn!(operation, error = %err, "authenticated PDS operation failed authorization");
        return unauthorized("authenticated PDS operation was not authorized");
    }
    tracing::error!(operation, error = %err, "authenticated PDS operation failed");
    internal("authenticated PDS operation failed")
}

fn client_error(operation: &str, err: jacquard_common::error::ClientError) -> AppError {
    let text = format!("{err:?}");
    if text.contains("Auth") || text.contains("Unauthorized") || text.contains("401") {
        tracing::warn!(operation, error = %err, "authenticated PDS read failed authorization");
        return unauthorized("authenticated PDS read was not authorized");
    }
    tracing::error!(operation, error = %err, "authenticated PDS read failed");
    internal("authenticated PDS read failed")
}

fn client_error_is_not_found(err: &jacquard_common::error::ClientError) -> bool {
    let text = format!("{err:?}");
    text.contains("RecordNotFound") || text.contains("not found") || text.contains("404")
}

fn agent_error_is_auth(err: &jacquard::client::AgentError) -> bool {
    format!("{err:?}").contains("Auth")
}

fn agent_error_is_not_found(err: &jacquard::client::AgentError) -> bool {
    let text = format!("{err:?}");
    text.contains("RecordNotFound") || text.contains("not found") || text.contains("404")
}

fn agent_error_is_conflict(err: &jacquard::client::AgentError) -> bool {
    let text = format!("{err:?}");
    text.contains("AlreadyExists") || text.contains("InvalidSwap") || text.contains("409")
}

#[cfg(test)]
mod tests {
    use std::str::FromStr;

    use jacquard_common::types::string::Did;
    use serde_json::json;
    use sqlx::SqlitePool;
    use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};

    use super::*;

    const DID_A: &str = "did:plc:aaaaaaaaaaaaaaaaaaaaaaaa";
    const DID_B: &str = "did:plc:bbbbbbbbbbbbbbbbbbbbbbbb";

    async fn state() -> AppState {
        let options = SqliteConnectOptions::from_str("sqlite::memory:").unwrap();
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect_with(options)
            .await
            .unwrap();
        sqlx::migrate!("./migrations").run(&pool).await.unwrap();
        let bootstrap = crate::oauth::bootstrap_oauth(pool.clone(), Some("http://localhost"))
            .expect("ephemeral OAuth bootstrap for tests");
        AppState::new(pool, bootstrap)
    }

    fn did(s: &str) -> Did {
        Did::new_owned(s).unwrap()
    }

    fn uri(did: &str, collection: &str, rkey: &str) -> String {
        format!("at://{did}/{collection}/{rkey}")
    }

    async fn seed_content_row(
        pool: &SqlitePool,
        table: &str,
        did: &str,
        rkey: &str,
        name: &str,
    ) -> String {
        let collection = match table {
            "things" => THING_COLLECTION,
            "models" => MODEL_COLLECTION,
            "parts" => PART_COLLECTION,
            _ => unreachable!(),
        };
        let uri = uri(did, collection, rkey);
        let now = chrono::Utc::now().timestamp_nanos_opt().unwrap();
        match table {
            "things" => {
                sqlx::query(
                    "INSERT INTO things (did, rkey, uri, cid, name, summary, license, tags_json, tags_text, instructions_text, cover_json, derived_from_uri, created_at, indexed_at, record_json) VALUES (?, ?, ?, 'bafycid', ?, NULL, 'CC-BY-4.0', NULL, NULL, NULL, NULL, NULL, ?, ?, ?)",
                )
                .bind(did)
                .bind(rkey)
                .bind(&uri)
                .bind(name)
                .bind(now)
                .bind(now)
                .bind(json!({"name": name, "license": "CC-BY-4.0", "models": []}).to_string())
                .execute(pool)
                .await
                .unwrap();
            }
            "models" => {
                sqlx::query(
                    "INSERT INTO models (did, rkey, uri, cid, name, summary, created_at, indexed_at, record_json) VALUES (?, ?, ?, 'bafycid', ?, NULL, ?, ?, ?)",
                )
                .bind(did)
                .bind(rkey)
                .bind(&uri)
                .bind(name)
                .bind(now)
                .bind(now)
                .bind(json!({"name": name, "parts": []}).to_string())
                .execute(pool)
                .await
                .unwrap();
            }
            "parts" => {
                let file_json =
                    json!({"mimeType": "model/stl", "size": 1, "chunks": []}).to_string();
                sqlx::query(
                    "INSERT INTO parts (did, rkey, uri, cid, name, format, file_json, created_at, indexed_at, record_json) VALUES (?, ?, ?, 'bafycid', ?, 'stl', ?, ?, ?, ?)",
                )
                .bind(did)
                .bind(rkey)
                .bind(&uri)
                .bind(name)
                .bind(&file_json)
                .bind(now)
                .bind(now)
                .bind(json!({"name": name, "file": {"mimeType": "model/stl", "size": 1, "chunks": []}}).to_string())
                .execute(pool)
                .await
                .unwrap();
            }
            _ => unreachable!(),
        }
        uri
    }

    #[tokio::test]
    async fn upload_staging_round_trips_owner_scoped_file_metadata() {
        let state = state().await;
        let actor = did(DID_A);
        let bytes = b"solid cube";
        let digest = hex_sha256(bytes);
        let file = File {
            chunks: Vec::new(),
            digest: Some(SmolStr::from(digest.clone())),
            mime_type: SmolStr::from("model/stl"),
            size: bytes.len() as i64,
            extra_data: None,
        };

        persist_upload(
            &state,
            &actor,
            UploadPersistence {
                upload_id: "upload-1",
                digest: &digest,
                mime_type: "model/stl",
                size: bytes.len(),
                filename: Some("cube.stl"),
                file: &file,
            },
        )
        .await
        .unwrap();

        let loaded = load_upload_file(&state, &actor, "upload-1").await.unwrap();
        assert_eq!(loaded.digest.as_deref(), Some(digest.as_str()));
        assert_eq!(loaded.mime_type.as_str(), "model/stl");
        assert_eq!(loaded.size, bytes.len() as i64);
        assert!(
            load_upload_file(&state, &did(DID_B), "upload-1")
                .await
                .is_err()
        );
    }

    #[tokio::test]
    async fn existing_like_lookup_returns_uri_and_cid_for_idempotent_duplicate_create() {
        let state = state().await;
        let actor = did(DID_A);
        let subject = seed_content_row(&state.pool, "things", DID_A, "thing1", "Thing").await;
        let cid = "bafyreiaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
        sqlx::query(
            "INSERT INTO likes (did, rkey, cid, subject_uri, created_at) VALUES (?, ?, ?, ?, 1)",
        )
        .bind(DID_A)
        .bind("like1")
        .bind(cid)
        .bind(&subject)
        .execute(&state.pool)
        .await
        .unwrap();

        let existing = lookup_existing_social(&state, &actor, LIKE_COLLECTION, &subject)
            .await
            .unwrap()
            .expect("like should be indexed");
        assert_eq!(existing.uri.as_ref(), uri(DID_A, LIKE_COLLECTION, "like1"));
        assert_eq!(existing.cid.to_string(), cid);
    }

    #[tokio::test]
    async fn subject_validation_accepts_indexed_content_and_rejects_unknowns() {
        let state = state().await;
        let thing = seed_content_row(&state.pool, "things", DID_A, "thing1", "Thing").await;
        let thing_uri = AtUri::new_owned(&thing).unwrap();
        validate_subject(&state, &thing_uri).await.unwrap();

        let missing = AtUri::new_owned(uri(DID_A, THING_COLLECTION, "missing")).unwrap();
        assert!(validate_subject(&state, &missing).await.is_err());
        let unsupported = AtUri::new_owned(uri(DID_A, LIKE_COLLECTION, "like1")).unwrap();
        assert!(validate_subject(&state, &unsupported).await.is_err());
    }

    #[tokio::test]
    async fn composition_load_requires_all_existing_children_to_be_owned() {
        let state = state().await;
        let thing = seed_content_row(&state.pool, "things", DID_A, "thing1", "Thing").await;
        let model = seed_content_row(&state.pool, "models", DID_A, "model1", "Model").await;
        let part = seed_content_row(&state.pool, "parts", DID_A, "part1", "Part").await;
        sqlx::query("INSERT INTO thing_models (thing_uri, model_uri, position) VALUES (?, ?, 0)")
            .bind(&thing)
            .bind(&model)
            .execute(&state.pool)
            .await
            .unwrap();
        sqlx::query("INSERT INTO model_parts (model_uri, part_uri, position) VALUES (?, ?, 0)")
            .bind(&model)
            .bind(&part)
            .execute(&state.pool)
            .await
            .unwrap();

        let composition =
            load_owned_composition(&state, &did(DID_A), &AtUri::new_owned(&thing).unwrap())
                .await
                .unwrap();
        assert_eq!(composition.models.len(), 1);
        assert_eq!(composition.parts.len(), 1);

        sqlx::query("UPDATE parts SET did = ? WHERE uri = ?")
            .bind(DID_B)
            .bind(&part)
            .execute(&state.pool)
            .await
            .unwrap();
        assert!(
            load_owned_composition(&state, &did(DID_A), &AtUri::new_owned(&thing).unwrap())
                .await
                .is_err()
        );
    }

    #[tokio::test]
    async fn update_child_validation_allows_owned_detached_records_to_be_reattached() {
        let state = state().await;
        let actor = did(DID_A);
        let thing = seed_content_row(&state.pool, "things", DID_A, "thing1", "Thing").await;
        let active_model = seed_content_row(&state.pool, "models", DID_A, "model1", "Model").await;
        let detached_model =
            seed_content_row(&state.pool, "models", DID_A, "model2", "Detached").await;
        let foreign_model =
            seed_content_row(&state.pool, "models", DID_B, "model3", "Foreign").await;
        sqlx::query("INSERT INTO thing_models (thing_uri, model_uri, position) VALUES (?, ?, 0)")
            .bind(&thing)
            .bind(&active_model)
            .execute(&state.pool)
            .await
            .unwrap();

        let previous = load_owned_composition(&state, &actor, &AtUri::new_owned(&thing).unwrap())
            .await
            .unwrap();
        let detached_uri = AtUri::new_owned(&detached_model).unwrap();
        let detached_rkey = rkey_from_uri(&detached_uri).unwrap();
        validate_existing_update_child(
            &state,
            &actor,
            &detached_uri,
            &detached_rkey,
            MODEL_COLLECTION,
            &previous.models,
        )
        .await
        .unwrap();

        let foreign_uri = AtUri::new_owned(&foreign_model).unwrap();
        let foreign_rkey = rkey_from_uri(&foreign_uri).unwrap();
        assert!(
            validate_existing_update_child(
                &state,
                &actor,
                &foreign_uri,
                &foreign_rkey,
                MODEL_COLLECTION,
                &previous.models,
            )
            .await
            .is_err()
        );
    }

    #[tokio::test]
    async fn child_reference_counts_identify_shared_delete_children() {
        let state = state().await;
        let thing_a = seed_content_row(&state.pool, "things", DID_A, "thing1", "Thing A").await;
        let thing_b = seed_content_row(&state.pool, "things", DID_A, "thing2", "Thing B").await;
        let model = seed_content_row(&state.pool, "models", DID_A, "model1", "Model").await;
        let part = seed_content_row(&state.pool, "parts", DID_A, "part1", "Part").await;
        sqlx::query("INSERT INTO thing_models (thing_uri, model_uri, position) VALUES (?, ?, 0)")
            .bind(&thing_a)
            .bind(&model)
            .execute(&state.pool)
            .await
            .unwrap();
        sqlx::query("INSERT INTO thing_models (thing_uri, model_uri, position) VALUES (?, ?, 0)")
            .bind(&thing_b)
            .bind(&model)
            .execute(&state.pool)
            .await
            .unwrap();
        sqlx::query("INSERT INTO model_parts (model_uri, part_uri, position) VALUES (?, ?, 0)")
            .bind(&model)
            .bind(&part)
            .execute(&state.pool)
            .await
            .unwrap();

        assert_eq!(
            model_reference_count(&state, &AtUri::new_owned(&model).unwrap())
                .await
                .unwrap(),
            2
        );
        assert_eq!(
            part_reference_count(&state, &AtUri::new_owned(&part).unwrap())
                .await
                .unwrap(),
            1
        );
    }

    #[tokio::test]
    async fn child_reference_counts_preserve_parts_reused_across_models() {
        let state = state().await;
        let thing = seed_content_row(&state.pool, "things", DID_A, "thing1", "Thing").await;
        let model_a = seed_content_row(&state.pool, "models", DID_A, "model1", "Model A").await;
        let model_b = seed_content_row(&state.pool, "models", DID_A, "model2", "Model B").await;
        let part = seed_content_row(&state.pool, "parts", DID_A, "part1", "Shared Part").await;
        for (position, model) in [(0, &model_a), (1, &model_b)] {
            sqlx::query(
                "INSERT INTO thing_models (thing_uri, model_uri, position) VALUES (?, ?, ?)",
            )
            .bind(&thing)
            .bind(model)
            .bind(position)
            .execute(&state.pool)
            .await
            .unwrap();
            sqlx::query("INSERT INTO model_parts (model_uri, part_uri, position) VALUES (?, ?, 0)")
                .bind(model)
                .bind(&part)
                .execute(&state.pool)
                .await
                .unwrap();
        }

        let composition =
            load_owned_composition(&state, &did(DID_A), &AtUri::new_owned(&thing).unwrap())
                .await
                .unwrap();
        assert_eq!(composition.models.len(), 2);
        assert_eq!(
            composition.parts.len(),
            1,
            "reused part should be deduped once"
        );
        assert_eq!(
            part_reference_count(&state, &AtUri::new_owned(&part).unwrap())
                .await
                .unwrap(),
            2
        );
    }
}
