//! Moving a mod to another account.
//!
//! A record can't change repos, so a transfer copies the listing, its media,
//! and its versions into the recipient's repo with their own session, copies
//! each deliverable inside the bucket, then hides the original. Deleting the
//! originals from the sender's repo needs the sender's session, which only
//! exists while they're signed in, so that waits for their next visit.

use anyhow::{anyhow, Context};
use atproto_record::lexicon::TypedBlob;
use axum::extract::{Form, Path, State};
use axum::response::{Html, IntoResponse, Redirect, Response};
use axum::routing::post;
use axum::Router;
use chrono::{DateTime, Duration, Utc};
use maud::{html, Markup};
use serde::Deserialize;
use sqlx::SqlitePool;

use crate::atproto::id::{AtUri, Did, Handle, Rkey};
use crate::atproto::lexicon::mod_listing::MediaItem;
use crate::atproto::lexicon::{ActorProfile, Found, ModListing, ModVersion, Record};
use crate::atproto::pds;
use crate::error::AppError;
use crate::form;
use crate::oauth::session::SessionCookie;
use crate::page::Page;
use crate::state::AppState;
use crate::{blobs, deliverables, s3};

/// How long an offer waits for its recipient.
const OFFER_DAYS: i64 = 30;

/// How long after an accept begins that it still counts as under way. One
/// that began longer ago than this and never finished has failed.
const ACCEPT_MINUTES: i64 = 15;

pub fn router() -> Router<AppState> {
    Router::new()
        .route("/{handle}/{slug}/transfer", post(handle_post_offer))
        .route("/{handle}/{slug}/transfer/check", post(handle_post_check))
        .route("/{handle}/{slug}/transfer/cancel", post(handle_post_cancel))
        .route("/transfers/{id}/accept", post(handle_post_accept))
        .route("/transfers/{id}/decline", post(handle_post_decline))
}

/// A pending offer of one of the sender's mods, as the sender sees it.
pub struct Outgoing {
    /// `@handle`, or the DID if the account has no handle.
    recipient: String,
    display_name: String,
    avatar_url: String,
    expires_at: DateTime<Utc>,
    /// Whether the recipient has started accepting, after which the offer
    /// can't be cancelled.
    accepting: bool,
    /// Whether an accept is under way, which makes the mod read-only until it
    /// finishes.
    pub in_progress: bool,
}

/// The pending offer of `listing`, if there is one.
pub async fn outgoing(
    state: &AppState,
    sender: &Did,
    listing: &Rkey,
) -> anyhow::Result<Option<Outgoing>> {
    let row = sqlx::query!(
        r#"SELECT recipient_did, expires_at, new_rkey IS NOT NULL AS "accepting!: bool"
           FROM mod_transfer
           WHERE sender_did = ? AND listing_rkey = ? AND accepted_at IS NULL
             AND (new_rkey IS NOT NULL OR expires_at > strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))"#,
        sender,
        listing
    )
    .fetch_optional(&state.db)
    .await?;
    let Some(row) = row else {
        return Ok(None);
    };

    let recipient = Did::new(row.recipient_did);
    let profile = ActorProfile::load(&state.db, &recipient).await?;
    Ok(Some(Outgoing {
        recipient: display_name(state, &recipient).await?,
        display_name: profile.display_name.clone(),
        avatar_url: profile.avatar_url(&recipient, 128),
        expires_at: DateTime::parse_from_rfc3339(&row.expires_at)?.with_timezone(&Utc),
        accepting: row.accepting,
        in_progress: accept_in_progress(&state.db, sender, listing).await?,
    }))
}

/// Whether `listing` is being copied to its new owner right now.
async fn accept_in_progress(db: &SqlitePool, sender: &Did, listing: &Rkey) -> sqlx::Result<bool> {
    let window = format!("-{ACCEPT_MINUTES} minutes");
    Ok(sqlx::query_scalar!(
        r#"SELECT 1 AS "in_progress!: i64" FROM mod_transfer
           WHERE sender_did = ? AND listing_rkey = ? AND accepted_at IS NULL
             AND accept_started_at > strftime('%Y-%m-%dT%H:%M:%fZ', 'now', ?)"#,
        sender,
        listing,
        window
    )
    .fetch_optional(db)
    .await?
    .is_some())
}

/// Fails if `listing` is being copied to its new owner, since a change made
/// now would be missing from the copy.
pub async fn ensure_not_transferring(
    state: &AppState,
    sender: &Did,
    listing: &Rkey,
) -> Result<(), AppError> {
    if accept_in_progress(&state.db, sender, listing)
        .await
        .map_err(|err| AppError::Internal(err.into()))?
    {
        return Err(AppError::BadRequest(
            "This mod is being transferred, so it can't be changed".to_string(),
        ));
    }
    Ok(())
}

/// How to name `did` to a person: `@handle`, or the DID if it has none.
async fn display_name(state: &AppState, did: &Did) -> anyhow::Result<String> {
    Ok(match did.resolve_handle(state).await? {
        Some(handle) => format!("@{handle}"),
        None => did.to_string(),
    })
}

/// Offers `listing` to `recipient`. Returns false if it's already offered to
/// someone.
async fn offer(
    db: &SqlitePool,
    sender: &Did,
    listing: &Rkey,
    slug: &str,
    recipient: &Did,
) -> sqlx::Result<bool> {
    let mut tx = db.begin().await?;
    // An expired offer would otherwise hold the mod's one slot.
    sqlx::query!(
        "DELETE FROM mod_transfer
         WHERE new_rkey IS NULL AND expires_at <= strftime('%Y-%m-%dT%H:%M:%fZ', 'now')"
    )
    .execute(&mut *tx)
    .await?;
    let expiry = format!("+{OFFER_DAYS} days");
    let inserted = sqlx::query!(
        "INSERT INTO mod_transfer
             (sender_did, listing_rkey, slug, recipient_did, created_at, expires_at)
         VALUES (?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%fZ', 'now'),
                 strftime('%Y-%m-%dT%H:%M:%fZ', 'now', ?))
         ON CONFLICT (sender_did, listing_rkey) DO NOTHING",
        sender,
        listing,
        slug,
        recipient,
        expiry
    )
    .execute(&mut *tx)
    .await?
    .rows_affected();
    tx.commit().await?;
    Ok(inserted == 1)
}

/// What the settings tab shows in place of its forms while the recipient's
/// accept is copying the mod. It polls, and leaves once the accept ends:
/// for the mod's new address if it finished, or the settings if it failed.
pub fn progress(existing: &Found<ModListing>, outgoing: &Outgoing) -> Markup {
    let settings = settings_path(existing);
    html! {
        div.transfer-progress role="status"
            hx-get=(settings) hx-trigger="every 3s"
            hx-target="#mod-details" hx-select="#mod-details" hx-swap="outerHTML"
            hx-on::response-error={ "window.location.href = '/" (existing.handle) "/" (existing.record.slug) "'" } {
            span.spinner aria-hidden="true" {}
            p.transfer-progress-heading { "Transferring to" }
            (recipient_chip(&outgoing.avatar_url, &outgoing.display_name, &outgoing.recipient))
        }
    }
}

/// An account's avatar and names, as `account` (`@handle`) with its display
/// name above it if it has one.
fn recipient_chip(avatar_url: &str, display_name: &str, account: &str) -> Markup {
    html! {
        div.transfer-recipient {
            img.transfer-avatar src=(avatar_url) alt="";
            @if display_name.is_empty() {
                strong { (account) }
            } @else {
                span.transfer-recipient-names {
                    strong { (display_name) }
                    span.transfer-recipient-handle { (account) }
                }
            }
        }
    }
}

/// The transfer section of a mod's settings: a button to offer it, or the
/// offer already pending and a button to cancel it.
pub fn section(existing: &Found<ModListing>, outgoing: Option<&Outgoing>) -> Markup {
    let base = format!("/{}/{}", existing.handle, existing.record.slug);
    html! {
        div.centered {
            @match outgoing {
                Some(outgoing) if outgoing.accepting => {
                    p { (outgoing.recipient) " is accepting this mod." }
                }
                Some(outgoing) => {
                    p {
                        "Waiting for " (outgoing.recipient) " to accept this mod. The offer expires on "
                        (outgoing.expires_at.format("%-d %B %Y")) "."
                    }
                    form method="post" action={ (base) "/transfer/cancel" } {
                        button.accent-red type="submit" { "Cancel transfer" }
                    }
                }
                None => {
                    div.label {
                        label { "Want someone else to look after this mod?" }
                    }
                    button.button type="button"
                        onclick="document.getElementById('transfer-dialog').showModal()" {
                        "Transfer"
                    }
                    dialog.upload-dialog id="transfer-dialog" aria-labelledby="transfer-dialog-title" {
                        h2 id="transfer-dialog-title" { "Transfer this mod" }
                        (find_form(&base, "", None, None))
                    }
                }
            }
        }
    }
}

/// An account found by handle, to confirm before offering it a mod.
struct Recipient {
    did: Did,
    handle: Handle,
    display_name: String,
    avatar_url: String,
}

/// The dialog's contents: a handle to look up and, once one is found, the
/// account it belongs to and a button to offer it the mod.
fn find_form(
    base: &str,
    value: &str,
    error: Option<&str>,
    recipient: Option<&Recipient>,
) -> Markup {
    html! {
        div id="transfer-find" {
            form method="post" action={ (base) "/transfer/check" }
                hx-post={ (base) "/transfer/check" } hx-target="#transfer-find" hx-swap="outerHTML"
                novalidate {
                p {
                    "The account you transfer this mod to gets its versions and downloads too, "
                    "and you can no longer edit it."
                }
                (form::text_field(
                    "recipient", "Handle", value,
                    Some(html! { "For example, alice.bsky.social." }),
                    error,
                ))
                div.form-actions {
                    button.button.secondary type="submit" { "Find account" }
                    button.secondary type="button" onclick="this.closest('dialog').close()" { "Cancel" }
                }
            }
            @if let Some(recipient) = recipient {
                form method="post" action={ (base) "/transfer" } {
                    input type="hidden" name="recipient" value=(recipient.did);
                    (recipient_chip(&recipient.avatar_url, &recipient.display_name, &format!("@{}", recipient.handle)))
                    div.form-actions {
                        button.button type="submit" { "Transfer to @" (recipient.handle) }
                    }
                }
            }
        }
    }
}

/// Finds the account a handle typed into the dialog belongs to, or says why
/// it can't have the mod.
async fn find_recipient(
    state: &AppState,
    sender: &Did,
    value: &str,
) -> Result<Recipient, &'static str> {
    let value = value.trim();
    if value.is_empty() {
        return Err("Enter a handle");
    }
    let handle = Handle::new(value).map_err(|_| "Enter a handle in the correct format")?;
    let did = match handle.resolve_did(state).await {
        Ok(Some(did)) => did,
        Ok(None) => return Err("Handle must belong to an existing account"),
        Err(err) => {
            eprintln!("failed to resolve {handle} for a transfer: {err:?}");
            return Err("The handle could not be checked - try again");
        }
    };
    if &did == sender {
        return Err("Enter the handle of another account");
    }
    let profile = ActorProfile::load(&state.db, &did).await.map_err(|err| {
        eprintln!("failed to load {did}'s profile for a transfer: {err:?}");
        "The handle could not be checked - try again"
    })?;
    Ok(Recipient {
        avatar_url: profile.avatar_url(&did, 128),
        display_name: profile.display_name,
        did,
        handle,
    })
}

#[derive(Deserialize)]
struct RecipientForm {
    recipient: String,
}

/// `POST /{handle}/{slug}/transfer/check` -- looks up the handle typed into
/// the transfer dialog, and shows whose account it is.
async fn handle_post_check(
    existing: Found<ModListing>,
    Form(form): Form<RecipientForm>,
) -> Result<Html<String>, AppError> {
    let session = existing.owner_session()?;
    let base = format!("/{}/{}", existing.handle, existing.record.slug);
    let markup = match find_recipient(existing.state(), &session.did, &form.recipient).await {
        Ok(recipient) => find_form(&base, &form.recipient, None, Some(&recipient)),
        Err(error) => find_form(&base, &form.recipient, Some(error), None),
    };
    Ok(Html(markup.into_string()))
}

/// `POST /{handle}/{slug}/transfer` -- offers the mod to the account the
/// dialog found.
async fn handle_post_offer(
    existing: Found<ModListing>,
    Form(form): Form<RecipientForm>,
) -> Result<Response, AppError> {
    let session = existing.owner_session()?;
    let recipient = Did::new(form.recipient);
    if recipient == session.did || !recipient.as_str().starts_with("did:") {
        return Err(AppError::BadRequest(
            "Enter the handle of another account".to_string(),
        ));
    }
    offer(
        &existing.state().db,
        &existing.did,
        &existing.rkey,
        &existing.record.slug,
        &recipient,
    )
    .await
    .map_err(|err| AppError::Internal(err.into()))?;
    Ok(Redirect::to(&settings_path(&existing)).into_response())
}

/// `POST /{handle}/{slug}/transfer/cancel` -- withdraws an offer nobody has
/// started accepting.
async fn handle_post_cancel(existing: Found<ModListing>) -> Result<Response, AppError> {
    existing.owner_session()?;
    sqlx::query!(
        "DELETE FROM mod_transfer
         WHERE sender_did = ? AND listing_rkey = ? AND new_rkey IS NULL",
        existing.did,
        existing.rkey
    )
    .execute(&existing.state().db)
    .await
    .map_err(|err| AppError::Internal(err.into()))?;
    Ok(Redirect::to(&settings_path(&existing)).into_response())
}

fn settings_path(existing: &Found<ModListing>) -> String {
    format!("/{}/{}/settings", existing.handle, existing.record.slug)
}

/// `POST /transfers/{id}/decline` -- refuses an offer, which forgets it.
async fn handle_post_decline(
    State(state): State<AppState>,
    session: SessionCookie,
    Path(id): Path<i64>,
) -> Result<Response, AppError> {
    sqlx::query!(
        "DELETE FROM mod_transfer WHERE id = ? AND recipient_did = ? AND new_rkey IS NULL",
        id,
        session.did
    )
    .execute(&state.db)
    .await
    .map_err(|err| AppError::Internal(err.into()))?;
    Ok(Redirect::to("/").into_response())
}

/// `POST /transfers/{id}/accept` -- moves the mod into the signed-in
/// recipient's repo.
async fn handle_post_accept(
    State(state): State<AppState>,
    session: SessionCookie,
    page: Page,
    Path(id): Path<i64>,
) -> Result<Response, AppError> {
    let accepted = match accept(&state, &session, id).await {
        Ok(accepted) => accepted,
        Err(AppError::Internal(err)) => {
            eprintln!("failed to accept transfer {id}: {err:?}");
            return Ok(page
                .body(html! {
                    main.form-page {
                        h1 { "The mod could not be transferred" }
                        p { "Something went wrong on our end. Select Accept to try again." }
                    }
                })
                .title("Transfer failed")
                .render()
                .into_response());
        }
        Err(err) => return Err(err),
    };

    let handle = session
        .did
        .resolve_handle(&state)
        .await?
        .ok_or_else(|| AppError::Internal(anyhow!("session has no resolvable handle")))?;
    let path = format!("/{handle}/{}", accepted.slug);
    if accepted.slug == accepted.old_slug {
        return Ok(Redirect::to(&path).into_response());
    }
    Ok(page
        .body(html! {
            main.form-page {
                h1 { (accepted.title) " is yours" }
                p {
                    "You already had a mod at /" (handle) "/" (accepted.old_slug)
                    ", so this one is at "
                    a href=(path) { (path) } "."
                }
            }
        })
        .title("Mod transferred")
        .render()
        .into_response())
}

/// A finished transfer: where the mod is now, and where it was.
struct Accepted {
    title: String,
    slug: String,
    old_slug: String,
}

/// Copies an offered mod into the recipient's repo, then hides the original.
///
/// Safe to run again after failing partway: the copy's rkeys and object keys
/// are recorded before anything is written to them, so a retry writes over
/// the same records and objects rather than making new ones. The original
/// stays listed until the copy is complete, so the mod always has an owner.
async fn accept(state: &AppState, session: &SessionCookie, id: i64) -> Result<Accepted, AppError> {
    let recipient = &session.did;
    let transfer = sqlx::query!(
        "SELECT sender_did, listing_rkey FROM mod_transfer
         WHERE id = ? AND recipient_did = ? AND accepted_at IS NULL
           AND (new_rkey IS NOT NULL OR expires_at > strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))",
        id,
        recipient
    )
    .fetch_optional(&state.db)
    .await
    .map_err(|err| AppError::Internal(err.into()))?
    .ok_or_else(|| AppError::BadRequest("The offer was cancelled or has expired".to_string()))?;
    let sender = Did::new(transfer.sender_did);
    let old_rkey = Rkey::from_verified(transfer.listing_rkey);
    sqlx::query!(
        "UPDATE mod_transfer SET accept_started_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
         WHERE id = ?",
        id
    )
    .execute(&state.db)
    .await
    .map_err(|err| AppError::Internal(err.into()))?;

    let Some(original) = ModListing::load_local(&state.db, &sender, &old_rkey).await? else {
        sqlx::query!("DELETE FROM mod_transfer WHERE id = ?", id)
            .execute(&state.db)
            .await
            .map_err(|err| AppError::Internal(err.into()))?;
        return Err(AppError::BadRequest(
            "The mod can't be transferred, because its owner deleted it".to_string(),
        ));
    };

    let (new_rkey, new_slug) = destination(&state.db, id, recipient, &original.slug).await?;

    // The same image uploaded twice is the same blob, so a retry costs a
    // transfer but makes nothing new.
    let mut media = Vec::with_capacity(original.media.len());
    for item in &original.media {
        let Some(file) = item.file() else {
            media.push(item.clone());
            continue;
        };
        media.push(MediaItem::Image {
            file: copy_blob(state, session, &sender, file).await?,
            alt: item.alt().map(str::to_string),
        });
    }
    let mut details_images = Vec::with_capacity(original.details_images.len());
    for file in &original.details_images {
        details_images.push(copy_blob(state, session, &sender, file).await?);
    }
    let copy = ModListing {
        slug: new_slug.clone(),
        media,
        details_images,
        ..original.clone()
    };
    copy.update(state, session, &new_rkey).await?;

    // Oldest first, so the copies' rkeys, minted in this order, sort the
    // same way the originals did.
    let mut versions = ModVersion::find_by_mod(&state.db, &sender, &old_rkey).await?;
    versions.reverse();
    for (old_version, version) in versions {
        copy_version(
            state,
            session,
            id,
            &sender,
            &new_rkey,
            &old_version,
            version,
        )
        .await?;
    }

    // One transaction, so the mod is never listed twice or not at all.
    let result: sqlx::Result<()> = async {
        let mut tx = state.db.begin().await?;
        sqlx::query!(
            "UPDATE mod_transfer SET accepted_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
             WHERE id = ?",
            id
        )
        .execute(&mut *tx)
        .await?;
        sqlx::query!(
            "DELETE FROM mod_listing WHERE author_did = ? AND rkey = ?",
            sender,
            old_rkey
        )
        .execute(&mut *tx)
        .await?;
        sqlx::query!(
            "DELETE FROM mod_version WHERE mod_did = ? AND mod_rkey = ?",
            sender,
            old_rkey
        )
        .execute(&mut *tx)
        .await?;
        sqlx::query!(
            "DELETE FROM mod_search WHERE author_did = ? AND rkey = ?",
            sender,
            old_rkey
        )
        .execute(&mut *tx)
        .await?;
        // The mod is as popular as it was. `WHERE true` tells SQLite the
        // ON CONFLICT is the upsert's, not a join's.
        sqlx::query!(
            "INSERT INTO mod_stats_day (mod_did, mod_rkey, day, downloads, views)
             SELECT ?, ?, day, downloads, views FROM mod_stats_day
             WHERE mod_did = ? AND mod_rkey = ? AND true
             ON CONFLICT (mod_did, mod_rkey, day)
             DO UPDATE SET downloads = downloads + excluded.downloads,
                           views = views + excluded.views",
            recipient,
            new_rkey,
            sender,
            old_rkey
        )
        .execute(&mut *tx)
        .await?;
        sqlx::query!(
            "DELETE FROM mod_stats_day WHERE mod_did = ? AND mod_rkey = ?",
            sender,
            old_rkey
        )
        .execute(&mut *tx)
        .await?;
        // Stars name the original, in their authors' repos, which only they
        // can change. Here, they count for the copy.
        sqlx::query!(
            "UPDATE OR IGNORE mod_star SET mod_did = ?, mod_rkey = ?
             WHERE mod_did = ? AND mod_rkey = ?",
            recipient,
            new_rkey,
            sender,
            old_rkey
        )
        .execute(&mut *tx)
        .await?;
        sqlx::query!(
            "UPDATE featured_mod SET mod_did = ?, mod_rkey = ?
             WHERE mod_did = ? AND mod_rkey = ?",
            recipient,
            new_rkey,
            sender,
            old_rkey
        )
        .execute(&mut *tx)
        .await?;
        tx.commit().await
    }
    .await;
    result.map_err(|err| AppError::Internal(err.into()))?;

    Ok(Accepted {
        title: original.title,
        slug: new_slug,
        old_slug: original.slug,
    })
}

/// Copies `sender`'s blob `file` to the PDS of `session`'s account and returns
/// the new ref.
async fn copy_blob(
    state: &AppState,
    session: &SessionCookie,
    sender: &Did,
    file: &TypedBlob,
) -> anyhow::Result<TypedBlob> {
    let blob = &file.inner;
    let (_, bytes) = blobs::fetch(state, sender.as_str(), &blob.ref_.link)
        .await?
        .ok_or_else(|| anyhow!("{sender}'s blob {} is gone", blob.ref_.link))?;
    pds::upload_blob(state, session, bytes, &blob.mime_type).await
}

/// Where transfer `id` copies its listing to in `recipient`'s repo: chosen
/// the first time it's accepted, and the same on every retry.
async fn destination(
    db: &SqlitePool,
    id: i64,
    recipient: &Did,
    slug: &str,
) -> anyhow::Result<(Rkey, String)> {
    let rkey = Rkey::new();
    let slug = free_slug(db, recipient, slug).await?;
    sqlx::query!(
        "UPDATE mod_transfer SET new_rkey = ?, new_slug = ? WHERE id = ? AND new_rkey IS NULL",
        rkey,
        slug,
        id
    )
    .execute(db)
    .await?;

    let row = sqlx::query!(
        r#"SELECT new_rkey AS "new_rkey!", new_slug AS "new_slug!" FROM mod_transfer WHERE id = ?"#,
        id
    )
    .fetch_one(db)
    .await?;
    Ok((Rkey::from_verified(row.new_rkey), row.new_slug))
}

/// `slug`, or the first of `slug-2`, `slug-3`, and so on that `did` has no
/// mod at.
async fn free_slug(db: &SqlitePool, did: &Did, slug: &str) -> anyhow::Result<String> {
    for n in 1.. {
        let candidate = suffixed(slug, n);
        let taken = sqlx::query_scalar!(
            r#"SELECT 1 AS "taken!: i64" FROM mod_listing WHERE author_did = ? AND slug = ?"#,
            did,
            candidate
        )
        .fetch_optional(db)
        .await?;
        if taken.is_none() {
            return Ok(candidate);
        }
    }
    unreachable!("the loop above only ends by returning")
}

/// The `n`th choice of slug: `slug` itself, then `slug-n`, shortened to fit
/// the lexicon's 64 bytes.
fn suffixed(slug: &str, n: u32) -> String {
    if n == 1 {
        return slug.to_string();
    }
    let suffix = format!("-{n}");
    let mut base = slug.to_string();
    while base.len() + suffix.len() > 64 {
        base.pop();
    }
    format!("{}{suffix}", base.trim_end_matches('-'))
}

/// Copies one version, and its deliverable, onto the copy of its listing.
async fn copy_version(
    state: &AppState,
    session: &SessionCookie,
    id: i64,
    sender: &Did,
    new_listing: &Rkey,
    old_rkey: &Rkey,
    version: ModVersion,
) -> anyhow::Result<()> {
    let recipient = &session.did;
    let old_key = sqlx::query_scalar!(
        "SELECT object_key FROM deliverable WHERE author_did = ? AND version_rkey = ?",
        sender,
        old_rkey
    )
    .fetch_optional(&state.db)
    .await?;

    let new_rkey = Rkey::new();
    let new_key = old_key.as_deref().map(|key| {
        let filename = key.rsplit('/').next().unwrap_or(key);
        s3::new_key(recipient, new_listing, filename)
    });
    sqlx::query!(
        "INSERT INTO mod_transfer_version (transfer_id, old_rkey, new_rkey, new_object_key)
         VALUES (?, ?, ?, ?)
         ON CONFLICT (transfer_id, old_rkey) DO NOTHING",
        id,
        old_rkey,
        new_rkey,
        new_key
    )
    .execute(&state.db)
    .await?;
    let row = sqlx::query!(
        "SELECT new_rkey, new_object_key FROM mod_transfer_version
         WHERE transfer_id = ? AND old_rkey = ?",
        id,
        old_rkey
    )
    .fetch_one(&state.db)
    .await?;
    let new_rkey = Rkey::from_verified(row.new_rkey);

    let deliverable = match (old_key, row.new_object_key) {
        (Some(old_key), Some(new_key)) => {
            state.s3.copy(&old_key, &new_key).await?;
            deliverables::record_checked(&state.db, &new_key, recipient, new_listing, &new_rkey)
                .await?;
            format!(
                "{}{}",
                state.config.external_base(),
                deliverables::download_path(recipient, new_listing, &new_rkey)
            )
        }
        // Not one of ours, so it's already somewhere anyone can get it.
        _ => version.deliverable.clone(),
    };

    let mut description_images = Vec::with_capacity(version.description_images.len());
    for file in &version.description_images {
        description_images.push(copy_blob(state, session, sender, file).await?);
    }
    let copy = ModVersion {
        listing: AtUri::new(recipient.clone(), ModListing::NSID, new_listing.clone()),
        deliverable,
        description_images,
        ..version
    };
    copy.update(state, session, &new_rkey)
        .await
        .map_err(|err| match err {
            crate::atproto::lexicon::WriteError::Invalid(err)
            | crate::atproto::lexicon::WriteError::Failed(err) => err,
        })
        .with_context(|| format!("failed to copy {sender}'s version {old_rkey}"))
}

/// Whether `did`'s listing at `rkey` was transferred away, so it must not be
/// listed again even while it's still in their repo.
pub async fn is_moved(db: &SqlitePool, did: &Did, rkey: &Rkey) -> anyhow::Result<bool> {
    let moved = sqlx::query_scalar!(
        r#"SELECT 1 AS "moved!: i64" FROM mod_transfer
           WHERE sender_did = ? AND listing_rkey = ? AND accepted_at IS NOT NULL"#,
        did,
        rkey
    )
    .fetch_optional(db)
    .await?;
    Ok(moved.is_some())
}

/// Where the mod `did` had at `slug` went: its new owner and slug.
pub async fn moved_listing(
    db: &SqlitePool,
    did: &Did,
    slug: &str,
) -> anyhow::Result<Option<(Did, String)>> {
    let row = sqlx::query!(
        r#"SELECT recipient_did, new_slug AS "new_slug!" FROM mod_transfer
           WHERE sender_did = ? AND slug = ? AND accepted_at IS NOT NULL
           ORDER BY accepted_at DESC LIMIT 1"#,
        did,
        slug
    )
    .fetch_optional(db)
    .await?;
    Ok(row.map(|row| (Did::new(row.recipient_did), row.new_slug)))
}

/// Where a transferred version's copy downloads from.
pub async fn moved_version(
    db: &SqlitePool,
    did: &Did,
    listing: &Rkey,
    version: &Rkey,
) -> anyhow::Result<Option<String>> {
    let row = sqlx::query!(
        r#"SELECT t.recipient_did, t.new_rkey AS "new_listing!", v.new_rkey AS new_version
           FROM mod_transfer t
           JOIN mod_transfer_version v ON v.transfer_id = t.id
           WHERE t.sender_did = ? AND t.listing_rkey = ? AND v.old_rkey = ?
             AND t.accepted_at IS NOT NULL"#,
        did,
        listing,
        version
    )
    .fetch_optional(db)
    .await?;
    Ok(row.map(|row| {
        deliverables::download_path(
            &Did::new(row.recipient_did),
            &Rkey::from_verified(row.new_listing),
            &Rkey::from_verified(row.new_version),
        )
    }))
}

/// An offer to show its recipient.
pub struct Banner {
    id: i64,
    sender: String,
    title: String,
    /// Whether accepting it started and failed partway, so it can only be
    /// tried again.
    accepting: bool,
}

/// The offers waiting for `recipient`.
pub async fn banners(state: &AppState, recipient: &Did) -> anyhow::Result<Vec<Banner>> {
    let rows = sqlx::query!(
        r#"SELECT t.id AS "id!", t.sender_did, l.title, t.new_rkey IS NOT NULL AS "accepting!: bool"
           FROM mod_transfer t
           JOIN mod_listing l ON l.author_did = t.sender_did AND l.rkey = t.listing_rkey
           WHERE t.recipient_did = ? AND t.accepted_at IS NULL
             AND (t.new_rkey IS NOT NULL OR t.expires_at > strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
           ORDER BY t.id"#,
        recipient
    )
    .fetch_all(&state.db)
    .await?;

    let mut banners = Vec::with_capacity(rows.len());
    for row in rows {
        banners.push(Banner {
            id: row.id,
            sender: display_name(state, &Did::new(row.sender_did)).await?,
            title: row.title,
            accepting: row.accepting,
        });
    }
    Ok(banners)
}

pub fn render_banners(banners: &[Banner]) -> Markup {
    html! {
        @if !banners.is_empty() {
            ul.transfer-offers.accent-blue aria-label="Mods offered to you" {
                @for banner in banners {
                    li {
                        span {
                            (banner.sender) " has requested to transfer "
                            strong { (banner.title) } " to you"
                        }
                        form method="post" action={ "/transfers/" (banner.id) "/accept" } {
                            button.button type="submit" { "Accept" }
                        }
                        @if !banner.accepting {
                            form method="post" action={ "/transfers/" (banner.id) "/decline" } {
                                button.secondary type="submit" { "Decline" }
                            }
                        }
                    }
                }
            }
        }
    }
}

/// Whether `sender` has transferred mods whose originals are still in their
/// repo.
pub async fn has_originals(db: &SqlitePool, sender: &Did) -> sqlx::Result<bool> {
    let pending = sqlx::query_scalar!(
        r#"SELECT 1 AS "pending!: i64" FROM mod_transfer
           WHERE sender_did = ? AND accepted_at IS NOT NULL AND originals_deleted_at IS NULL
           LIMIT 1"#,
        sender
    )
    .fetch_optional(db)
    .await?;
    Ok(pending.is_some())
}

/// Deletes the originals of the mods `session`'s account transferred away,
/// from their repo and the bucket. Each is marked done only once all of it
/// is gone, so a failure is retried the next time they visit.
pub async fn delete_originals(state: &AppState, session: &SessionCookie) -> anyhow::Result<()> {
    let sender = &session.did;
    let transfers = sqlx::query!(
        r#"SELECT id AS "id!", listing_rkey FROM mod_transfer
           WHERE sender_did = ? AND accepted_at IS NOT NULL AND originals_deleted_at IS NULL"#,
        sender
    )
    .fetch_all(&state.db)
    .await?;

    for transfer in transfers {
        let versions = sqlx::query_scalar!(
            "SELECT old_rkey FROM mod_transfer_version WHERE transfer_id = ?",
            transfer.id
        )
        .fetch_all(&state.db)
        .await?;
        for version in versions {
            pds::delete(state, session, ModVersion::NSID, &version).await?;
        }
        pds::delete(state, session, ModListing::NSID, &transfer.listing_rkey).await?;

        let listing = Rkey::from_verified(transfer.listing_rkey);
        let prefix = format!("{sender}/{listing}/");
        state.s3.delete_prefix(&prefix).await?;
        sqlx::query!(
            "DELETE FROM deliverable WHERE author_did = ? AND mod_rkey = ?",
            sender,
            listing
        )
        .execute(&state.db)
        .await?;

        sqlx::query!(
            "UPDATE mod_transfer SET originals_deleted_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
             WHERE id = ?",
            transfer.id
        )
        .execute(&state.db)
        .await?;
    }
    Ok(())
}

/// Starts deleting the originals of `session`'s transferred mods, if any
/// are left and its token will last long enough to.
pub async fn spawn_delete_originals(state: &AppState, session: SessionCookie) {
    if session.expires_within(Duration::minutes(1)) {
        return;
    }
    match has_originals(&state.db, &session.did).await {
        Ok(false) => {}
        Ok(true) => {
            let state = state.clone();
            tokio::spawn(async move {
                if let Err(err) = delete_originals(&state, &session).await {
                    eprintln!(
                        "failed to delete {}'s transferred originals: {err:?}",
                        session.did
                    );
                }
            });
        }
        Err(err) => eprintln!(
            "failed to check {} for transferred originals: {err:?}",
            session.did
        ),
    }
}

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

    async fn test_db() -> SqlitePool {
        let db = SqlitePool::connect("sqlite::memory:").await.unwrap();
        sqlx::migrate!().run(&db).await.unwrap();
        db
    }

    #[test]
    fn a_taken_slug_gets_a_number_and_still_fits() {
        assert_eq!(suffixed("master-quest", 1), "master-quest");
        assert_eq!(suffixed("master-quest", 2), "master-quest-2");

        let long = "a".repeat(64);
        let shortened = suffixed(&long, 12);
        assert_eq!(shortened.len(), 64);
        assert!(shortened.ends_with("a-12"));
    }

    #[tokio::test]
    async fn a_mod_can_only_be_offered_to_one_account_at_a_time() {
        let db = test_db().await;
        let sender = Did::new("did:plc:sender");
        let listing = Rkey::new();

        assert!(
            offer(&db, &sender, &listing, "mq", &Did::new("did:plc:bob"))
                .await
                .unwrap()
        );
        assert!(
            !offer(&db, &sender, &listing, "mq", &Did::new("did:plc:carol"))
                .await
                .unwrap()
        );
    }

    #[tokio::test]
    async fn an_expired_offer_makes_way_for_a_new_one() {
        let db = test_db().await;
        let sender = Did::new("did:plc:sender");
        let listing = Rkey::new();

        offer(&db, &sender, &listing, "mq", &Did::new("did:plc:bob"))
            .await
            .unwrap();
        sqlx::query("UPDATE mod_transfer SET expires_at = '2000-01-01T00:00:00.000Z'")
            .execute(&db)
            .await
            .unwrap();

        assert!(
            offer(&db, &sender, &listing, "mq", &Did::new("did:plc:carol"))
                .await
                .unwrap()
        );
        let recipient: String = sqlx::query_scalar("SELECT recipient_did FROM mod_transfer")
            .fetch_one(&db)
            .await
            .unwrap();
        assert_eq!(recipient, "did:plc:carol");
    }

    #[tokio::test]
    async fn an_accept_that_began_long_ago_no_longer_locks_the_mod() {
        let db = test_db().await;
        let sender = Did::new("did:plc:sender");
        let listing = Rkey::new();
        offer(&db, &sender, &listing, "mq", &Did::new("did:plc:bob"))
            .await
            .unwrap();
        assert!(!accept_in_progress(&db, &sender, &listing).await.unwrap());

        sqlx::query(
            "UPDATE mod_transfer SET accept_started_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now')",
        )
        .execute(&db)
        .await
        .unwrap();
        assert!(accept_in_progress(&db, &sender, &listing).await.unwrap());

        sqlx::query("UPDATE mod_transfer SET accept_started_at = '2000-01-01T00:00:00.000Z'")
            .execute(&db)
            .await
            .unwrap();
        assert!(!accept_in_progress(&db, &sender, &listing).await.unwrap());
    }

    #[tokio::test]
    async fn the_destination_is_chosen_once_and_skips_taken_slugs() {
        let db = test_db().await;
        let sender = Did::new("did:plc:sender");
        let recipient = Did::new("did:plc:bob");
        offer(&db, &sender, &Rkey::new(), "mq", &recipient)
            .await
            .unwrap();
        let id: i64 = sqlx::query_scalar("SELECT id FROM mod_transfer")
            .fetch_one(&db)
            .await
            .unwrap();
        sqlx::query(
            "INSERT INTO mod_listing (author_did, rkey, title, slug, description, game, created_at)
             VALUES ('did:plc:bob', '3kaaa', 'Mine', 'mq', '', '64', '2026-01-01T00:00:00Z')",
        )
        .execute(&db)
        .await
        .unwrap();

        let first = destination(&db, id, &recipient, "mq").await.unwrap();
        assert_eq!(first.1, "mq-2");
        let retried = destination(&db, id, &recipient, "mq").await.unwrap();
        assert_eq!(retried, first);
    }

    #[tokio::test]
    async fn only_an_accepted_transfer_moves_the_old_address() {
        let db = test_db().await;
        let sender = Did::new("did:plc:sender");
        let recipient = Did::new("did:plc:bob");
        let listing = Rkey::new();
        offer(&db, &sender, &listing, "mq", &recipient)
            .await
            .unwrap();
        let id: i64 = sqlx::query_scalar("SELECT id FROM mod_transfer")
            .fetch_one(&db)
            .await
            .unwrap();
        destination(&db, id, &recipient, "mq").await.unwrap();

        assert!(!is_moved(&db, &sender, &listing).await.unwrap());
        assert!(moved_listing(&db, &sender, "mq").await.unwrap().is_none());

        sqlx::query("UPDATE mod_transfer SET accepted_at = '2026-10-11T00:00:00.000Z'")
            .execute(&db)
            .await
            .unwrap();
        assert!(is_moved(&db, &sender, &listing).await.unwrap());
        assert_eq!(
            moved_listing(&db, &sender, "mq").await.unwrap(),
            Some((recipient, "mq".to_string()))
        );
    }
}