//! Shared playlists (F3 → the app): a playlist is published as an
//! immutable, content-addressed REVISION (the portable manifest — entry
//! occurrences with display fallbacks, never a path) plus an owner-signed
//! POINTER to the current revision per (owner, playlist uuid). Editing
//! publishes a new revision and bumps the pointer; withdrawing bumps the
//! pointer to "none", and the truthful line stays: published revisions
//! remain fetchable, nothing on the network is erased.
//!
//! Both contract wasms are bundled commitments like the persona ones
//! (`assets/playlist_revision_contract.wasm`, `assets/playlist_pointer_
//! contract.wasm`, from research/freenet-f3; code hashes pinned by tests).
//!
//! Sharing WITH someone sends an address, nothing else — `SSSHARE1?<ref>`
//! over the same Soulseek PM channel the link proof and friends use. The
//! recipient sees a card, fetches the manifest through their own node,
//! verifies it against the owner's key, and imports it through the app's
//! normal playlist-import path (the review resolves tracks against their
//! library and the catalog, and unowned rows become Soulseek streams).

use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::path::PathBuf;
use std::sync::{Mutex, OnceLock};
use std::time::{SystemTime, UNIX_EPOCH};

use freenet_adapter::FreenetAdapter;
use library::NewPlaylistItem;
use persona_state as ps;
use persona_state::playlist as pl;
use persona_state::slsk_link::normalize_username;
use serde::{Deserialize, Serialize};
use tauri::{Emitter, Manager, State};

use crate::inbox::{self, Cargo, ShareCargo};
use crate::ledger as ledger_lock;
use crate::persona;
use crate::pipeline::LibraryState;
use crate::playlist_formats::{ParsedEntry, ParsedPlaylist};
use crate::playlists::{self, DraftEntry, ImportDraft};
use crate::state::AppState;

const REV_WASM: &[u8] = include_bytes!("../assets/playlist_revision_contract.wasm");
const PTR_WASM: &[u8] = include_bytes!("../assets/playlist_pointer_contract.wasm");
const SHARE_PREFIX: &str = "SSSHARE1?";
pub const EVENT: &str = "community:shares";

/// A playlist this device has published, by local playlist id.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Published {
    pub playlist_id: i64,
    pub uuid: String,
    pub revision_seq: u64,
    pub pointer_seq: u64,
    /// The current revision's content hash (b58); stays after a withdraw so
    /// the next publication still links to its predecessor.
    pub current_hash: Option<String>,
    pub name: String,
    pub entry_count: u32,
    pub withdrawn: bool,
    pub published_at: u64,
}

/// A reference someone sent us: an address, plus what they said it was.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ShareCard {
    pub id: String,
    /// Empty when it arrived sealed from a persona this device has no
    /// Soulseek account for; `from_key` is then the only name.
    pub from_username: String,
    #[serde(default)]
    pub from_key: String,
    /// "pm" | "inbox" — how it arrived.
    #[serde(default = "carry_pm")]
    pub carry: String,
    pub owner_key: String,
    pub playlist_uuid: String,
    pub name: String,
    pub entry_count: u32,
    pub received_at: u64,
}

fn carry_pm() -> String {
    "pm".into()
}

#[derive(Default, Serialize, Deserialize)]
struct Ledger {
    #[serde(default)]
    published: BTreeMap<String, Published>,
    #[serde(default)]
    inbox: Vec<ShareCard>,
    /// Shared playlists this device imported and keeps an eye on, keyed
    /// `owner:uuid` — the one key a database cannot hand out twice.
    #[serde(default)]
    following: BTreeMap<String, Follow>,
}

/// A shared playlist this device imported and follows. Importing is a
/// snapshot; following is remembering WHICH snapshot, so the owner's next
/// revision reads as "3 new tracks since you imported" rather than as a
/// whole playlist again. One follow per shared playlist: importing it a
/// second time moves the follow to the new copy.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Follow {
    pub owner_key: String,
    pub playlist_uuid: String,
    /// The local playlist the import made. Deleting it ends the follow —
    /// SQLite hands the row id to the next playlist created, and nothing
    /// may keep pointing at it (`forget_playlist`).
    pub playlist_id: i64,
    pub name: String,
    /// The revision the local copy last took, by content hash.
    pub synced_hash: String,
    pub synced_seq: u64,
    pub synced_at: u64,
    pub followed_at: u64,
    /// The last check: when it ran and the revision the pointer named.
    /// `seen_hash` is empty while the owner has the playlist withdrawn.
    pub checked_at: u64,
    pub seen_hash: String,
    pub seen_seq: u64,
    pub withdrawn: bool,
    /// Entries `seen` has that `synced` never had, in the owner's order, and
    /// how many of `synced`'s entries the owner has since dropped.
    #[serde(default)]
    pub added: Vec<ShareEntry>,
    #[serde(default)]
    pub removed: u32,
    /// Why the last check could not read the pointer, if it could not.
    #[serde(default)]
    pub last_error: Option<String>,
}

/// What travels in a share: the address and a display hint.
#[derive(Serialize, Deserialize)]
struct ShareRef {
    owner: [u8; 32],
    playlist_uuid: [u8; 16],
    name: String,
    entry_count: u32,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ShareEntry {
    /// The occurrence id (b58) — what a proposal's remove or move names.
    pub entry_id: String,
    pub position: u32,
    pub title: String,
    pub artist: String,
    pub album: String,
    pub duration_secs: Option<u32>,
    pub note: Option<String>,
}

/// A fetched, VERIFIED shared playlist as the reader sees it.
#[derive(Debug, Serialize)]
pub struct ShareView {
    pub owner_key: String,
    pub playlist_uuid: String,
    pub name: String,
    pub description: String,
    pub revision_seq: u64,
    pub content_hash: String,
    pub withdrawn: bool,
    pub entries: Vec<ShareEntry>,
}

fn now_secs() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}
fn rand16() -> [u8; 16] {
    use rand::RngCore;
    let mut n = [0u8; 16];
    rand::thread_rng().fill_bytes(&mut n);
    n
}
fn b58(bytes: &[u8]) -> String {
    bs58::encode(bytes).into_string()
}
fn bytes32(b58s: &str) -> Result<[u8; 32], String> {
    bs58::decode(b58s)
        .into_vec()
        .map_err(|e| format!("bad key: {e}"))?
        .as_slice()
        .try_into()
        .map_err(|_| "not 32 bytes".to_string())
}
fn bytes16(b58s: &str) -> Result<[u8; 16], String> {
    bs58::decode(b58s)
        .into_vec()
        .map_err(|e| format!("bad id: {e}"))?
        .as_slice()
        .try_into()
        .map_err(|_| "not 16 bytes".to_string())
}
/// Cut to a byte cap on a char boundary.
fn cap(s: &str, max: usize) -> String {
    if s.len() <= max {
        return s.to_string();
    }
    let mut end = max;
    while end > 0 && !s.is_char_boundary(end) {
        end -= 1;
    }
    s[..end].to_string()
}

fn path(app: &tauri::AppHandle) -> Result<PathBuf, String> {
    Ok(crate::paths::app_data(app)
        .map_err(|e| e.to_string())?
        .join("freenet")
        .join(LEDGER_FILE))
}
/// Both the file the ledger lives in and the key its write lock is taken on.
const LEDGER_FILE: &str = "playlists-ledger.json";

fn load(app: &tauri::AppHandle) -> Ledger {
    path(app)
        .ok()
        .and_then(|p| std::fs::read(p).ok())
        .and_then(|b| serde_json::from_slice(&b).ok())
        .unwrap_or_default()
}
fn save(app: &tauri::AppHandle, ledger: &Ledger) -> Result<(), String> {
    let p = path(app)?;
    if let Some(dir) = p.parent() {
        std::fs::create_dir_all(dir).map_err(|e| e.to_string())?;
    }
    std::fs::write(&p, serde_json::to_vec_pretty(ledger).map_err(|e| e.to_string())?)
        .map_err(|e| e.to_string())?;
    let _ = app.emit(EVENT, ());
    Ok(())
}

async fn own_key(app: &tauri::AppHandle) -> Result<[u8; 32], String> {
    let adapter = app.state::<AppState>().freenet.clone();
    let info = persona::current(&adapter)
        .await?
        .ok_or("create a SoulStream identity first")?;
    bytes32(&info.public_key)
}
async fn send_pm(app: &tauri::AppHandle, to: &str, text: &str) -> Result<(), String> {
    let st = app.state::<AppState>();
    let guard = st.client.lock().await;
    guard
        .as_ref()
        .ok_or("not connected to Soulseek")?
        .message_user(to, text)
        .map_err(|e| e.to_string())
}

// ---------------------------------------------------------------- publish

/// Publish (or re-publish) a local playlist: a new signed revision and a
/// pointer bump. Nothing about local files leaves the machine — entries
/// carry titles, artists, albums, durations and a text hint only.
pub async fn publish(app: &tauri::AppHandle, lib: &LibraryState, playlist_id: i64) -> Result<Published, String> {
    let adapter: FreenetAdapter = app.state::<AppState>().freenet.clone();
    let owner = own_key(app).await?;
    let l = lib.library.clone();
    let (meta, items) = tauri::async_runtime::spawn_blocking(move || -> Result<_, String> {
        let meta = l
            .list_playlists()
            .map_err(|e| e.to_string())?
            .into_iter()
            .find(|p| p.id == playlist_id)
            .ok_or("no such playlist")?;
        let items = l.playlist_items(playlist_id).map_err(|e| e.to_string())?;
        Ok((meta, items))
    })
    .await
    .map_err(|e| e.to_string())??;
    if items.len() > pl::MAX_ENTRIES {
        return Err(format!("{} entries is over the {} cap", items.len(), pl::MAX_ENTRIES));
    }
    let _guard = ledger_lock::hold(LEDGER_FILE).await;
    let mut ledger = load(app);
    let key = playlist_id.to_string();
    let prev = ledger.published.get(&key).cloned();
    let uuid = match &prev {
        Some(p) => bytes16(&p.uuid)?,
        None => rand16(),
    };
    let revision_seq = prev.as_ref().map(|p| p.revision_seq + 1).unwrap_or(1);
    let prev_hash = match prev.as_ref().and_then(|p| p.current_hash.as_deref()) {
        Some(h) => Some(bytes32(h)?),
        None => None,
    };
    let entries: Vec<pl::EntryV1> = items
        .iter()
        .enumerate()
        .map(|(i, it)| pl::EntryV1 {
            entry_id: entry_id_for(&uuid, it.id),
            position: i as u32,
            recording: pl::RecordingRef {
                musicbrainz_recording_id: None,
                isrcs: vec![],
                deezer_id: it.track_id.clone().filter(|t| !t.is_empty()),
                apple_music_id: None,
                title: cap(&it.title, pl::MAX_FIELD_BYTES),
                artists: vec![cap(&it.artist, pl::MAX_FIELD_BYTES)],
                album: cap(it.album.as_deref().unwrap_or(""), pl::MAX_FIELD_BYTES),
                duration_ms: it.duration_secs.map(|s| s.saturating_mul(1000)),
                release_year: None,
            },
            note: it.note.as_deref().map(|n| cap(n, pl::MAX_NOTE_BYTES)),
            slsk_hint: Some(cap(&format!("{} {}", it.artist, it.title), pl::MAX_FIELD_BYTES)),
        })
        .collect();
    let mut m = pl::ManifestV1 {
        schema_version: pl::SCHEMA_VERSION,
        owner,
        playlist_uuid: uuid,
        revision_seq,
        prev_revision_hash: prev_hash,
        name: cap(&meta.name, pl::MAX_NAME_BYTES),
        description: cap(meta.description.as_deref().unwrap_or(""), pl::MAX_DESC_BYTES),
        artwork: None,
        entries,
        sig: vec![],
    };
    m.sig = persona::sign(&adapter, pl::SIGN_DOMAIN, pl::manifest_sign_payload(&m)).await?;
    let bytes = bincode::serialize(&m).map_err(|e| e.to_string())?;
    let hash = ps::content_hash(&bytes);
    adapter
        .contract_put(REV_WASM.to_vec(), &pl::make_rev_params(&owner, &hash), bytes.clone(), false)
        .await
        .map_err(|e| e.to_string())?;
    cache_revision(app, &hash, &bytes);
    let pointer_seq = prev.as_ref().map(|p| p.pointer_seq + 1).unwrap_or(1);
    let mut p = pl::PointerState {
        schema_version: pl::SCHEMA_VERSION,
        owner,
        playlist_uuid: uuid,
        sequence: pointer_seq,
        current: Some(pl::RevisionRef {
            content_hash: hash,
            revision_seq,
            entry_count: m.entries.len() as u32,
            byte_length: bytes.len() as u32,
            name: m.name.clone(),
        }),
        withdrawn: false,
        sig: vec![],
    };
    p.sig = persona::sign(&adapter, pl::SIGN_DOMAIN, pl::pointer_sign_payload(&p)).await?;
    let pb = bincode::serialize(&p).map_err(|e| e.to_string())?;
    let params = pl::make_ptr_params(&owner, &uuid);
    if prev.is_some() {
        adapter.contract_update(PTR_WASM.to_vec(), &params, pb).await.map(|_| ())
    } else {
        adapter.contract_put(PTR_WASM.to_vec(), &params, pb, false).await.map(|_| ())
    }
    .map_err(|e| e.to_string())?;
    let rec = Published {
        playlist_id,
        uuid: b58(&uuid),
        revision_seq,
        pointer_seq,
        current_hash: Some(b58(&hash)),
        name: m.name.clone(),
        entry_count: m.entries.len() as u32,
        withdrawn: false,
        published_at: now_secs(),
    };
    ledger.published.insert(key, rec.clone());
    save(app, &ledger)?;
    log::info!("shared playlists: published {:?} rev {} ({} entries)", rec.name, revision_seq, rec.entry_count);
    Ok(rec)
}

/// Withdraw: the pointer moves to "none". Revisions already published stay
/// fetchable on the network — the ledger keeps saying so.
pub async fn withdraw(app: &tauri::AppHandle, playlist_id: i64) -> Result<Published, String> {
    let adapter = app.state::<AppState>().freenet.clone();
    let owner = own_key(app).await?;
    let _guard = ledger_lock::hold(LEDGER_FILE).await;
    let mut ledger = load(app);
    let key = playlist_id.to_string();
    let prev = ledger.published.get(&key).cloned().ok_or("that playlist was never published")?;
    if prev.withdrawn {
        return Ok(prev);
    }
    let uuid = bytes16(&prev.uuid)?;
    let mut p = pl::PointerState {
        schema_version: pl::SCHEMA_VERSION,
        owner,
        playlist_uuid: uuid,
        sequence: prev.pointer_seq + 1,
        current: None,
        withdrawn: true,
        sig: vec![],
    };
    p.sig = persona::sign(&adapter, pl::SIGN_DOMAIN, pl::pointer_sign_payload(&p)).await?;
    let pb = bincode::serialize(&p).map_err(|e| e.to_string())?;
    adapter
        .contract_update(PTR_WASM.to_vec(), &pl::make_ptr_params(&owner, &uuid), pb)
        .await
        .map_err(|e| e.to_string())?;
    let rec = Published { pointer_seq: prev.pointer_seq + 1, withdrawn: true, ..prev };
    ledger.published.insert(key, rec.clone());
    save(app, &ledger)?;
    Ok(rec)
}

// ---------------------------------------------------------------- read

/// Resolve (owner, uuid) through the network: pointer, then the revision it
/// names, verified in the plan's order before anything is shown.
pub async fn fetch(app: &tauri::AppHandle, owner_b58: &str, uuid_b58: &str) -> Result<ShareView, String> {
    let adapter = app.state::<AppState>().freenet.clone();
    let owner = bytes32(owner_b58)?;
    let uuid = bytes16(uuid_b58)?;
    let ptr_ids = FreenetAdapter::contract_ids(PTR_WASM.to_vec(), &pl::make_ptr_params(&owner, &uuid));
    let pbytes = adapter.contract_get(&ptr_ids.instance, false).await.map_err(|e| {
        if e.to_string().contains("not found") {
            "no such shared playlist on the network (or not reachable yet)".to_string()
        } else {
            e.to_string()
        }
    })?;
    let p = pl::validate_pointer(&owner, &uuid, &pbytes).map_err(|e| format!("pointer failed verification: {e}"))?;
    let Some(r) = p.current.filter(|_| !p.withdrawn) else {
        return Ok(ShareView {
            owner_key: owner_b58.to_string(),
            playlist_uuid: uuid_b58.to_string(),
            name: String::new(),
            description: String::new(),
            revision_seq: 0,
            content_hash: String::new(),
            withdrawn: true,
            entries: vec![],
        });
    };
    let (m, _) = revision_by_hash(app, &owner, &r.content_hash).await?;
    Ok(ShareView {
        owner_key: owner_b58.to_string(),
        playlist_uuid: uuid_b58.to_string(),
        name: m.name,
        description: m.description,
        revision_seq: m.revision_seq,
        content_hash: b58(&r.content_hash),
        withdrawn: false,
        entries: m.entries.iter().map(share_entry).collect(),
    })
}

/// A manifest entry as the reader sees it: display fields and the
/// occurrence id, never the owner's hints.
fn share_entry(e: &pl::EntryV1) -> ShareEntry {
    ShareEntry {
        entry_id: b58(&e.entry_id),
        position: e.position,
        title: e.recording.title.clone(),
        artist: e.recording.artists.join(", "),
        album: e.recording.album.clone(),
        duration_secs: e.recording.duration_ms.map(|ms| ms / 1000),
        note: e.note.clone(),
    }
}

/// Shared entries as the import doors take them, so a shared playlist —
/// or the new rows of one — resolves exactly the way a pasted list does.
fn parsed_from(name: &str, description: &str, entries: &[ShareEntry]) -> ParsedPlaylist {
    ParsedPlaylist {
        format: "soulstream".into(),
        title: Some(name.to_string()),
        description: Some(description.to_string()).filter(|d| !d.trim().is_empty()),
        entries: entries
            .iter()
            .map(|e| ParsedEntry {
                ordinal: e.position + 1,
                title: Some(e.title.clone()),
                artist: Some(e.artist.clone()),
                album: Some(e.album.clone()).filter(|a| !a.trim().is_empty()),
                duration_secs: e.duration_secs,
                location: None,
                start_ms: None,
                raw: format!("{} - {}", e.artist, e.title),
            })
            .collect(),
        warnings: vec![],
    }
}

/// A resolved draft row as a playlist row — `playlist_import_commit`'s own
/// mapping, minus the setlist segment case a shared playlist never carries.
/// An empty kind lets the insert classify: a matched path is local, a full
/// identity streams, anything weaker stays an unresolved placeholder.
fn new_row(e: DraftEntry) -> NewPlaylistItem {
    NewPlaylistItem {
        title: e.title,
        artist: e.artist,
        album: e.album,
        duration_secs: e.duration_secs,
        kind: String::new(),
        path: e.path.filter(|p| !p.is_empty()),
        art_path: e.art_path,
        source_line: (!e.raw.is_empty()).then_some(e.raw),
        start_ms: e.start_ms,
        note: None,
        track_id: e.track_id.map(|id| id.to_string()),
        art_url: e.art_url,
    }
}

/// The occurrence id a local row publishes under: derived from the playlist
/// uuid and the row id, so it survives re-publication — a proposal's remove
/// or move names the same row the owner has.
pub(crate) fn entry_id_for(uuid: &[u8; 16], item_id: i64) -> [u8; 16] {
    let mut buf = Vec::with_capacity(48);
    buf.extend_from_slice(b"soulstream.entry.v1");
    buf.extend_from_slice(uuid);
    buf.extend_from_slice(&item_id.to_le_bytes());
    ps::content_hash(&buf)[..16].try_into().unwrap()
}

fn revisions_dir(app: &tauri::AppHandle) -> Option<PathBuf> {
    crate::paths::app_data(app).ok().map(|d| d.join("freenet").join("revisions"))
}

/// Revisions are kept on disk by hash: the owner's accept path needs the
/// exact base bytes, and a network GET for them costs seconds.
fn cache_revision(app: &tauri::AppHandle, hash: &[u8; 32], bytes: &[u8]) {
    if let Some(dir) = revisions_dir(app) {
        let _ = std::fs::create_dir_all(&dir);
        let _ = std::fs::write(dir.join(format!("{}.bin", b58(hash))), bytes);
    }
}

/// A revision by content hash — the local cache first, the network after —
/// verified against the owner's key either way.
pub(crate) async fn revision_by_hash(
    app: &tauri::AppHandle,
    owner: &[u8; 32],
    hash: &[u8; 32],
) -> Result<(pl::ManifestV1, Vec<u8>), String> {
    let cached = revisions_dir(app).and_then(|d| std::fs::read(d.join(format!("{}.bin", b58(hash)))).ok());
    let bytes = match cached {
        Some(b) => b,
        None => {
            let adapter = app.state::<AppState>().freenet.clone();
            let ids = FreenetAdapter::contract_ids(REV_WASM.to_vec(), &pl::make_rev_params(owner, hash));
            let b = adapter.contract_get(&ids.instance, false).await.map_err(|e| e.to_string())?;
            cache_revision(app, hash, &b);
            b
        }
    };
    let m = pl::validate_revision(owner, hash, &bytes).map_err(|e| format!("revision failed verification: {e}"))?;
    Ok((m, bytes))
}

/// The current revision of (owner, uuid): the pointer, then the revision it
/// names. None when the pointer says withdrawn.
pub(crate) async fn current_revision(
    app: &tauri::AppHandle,
    owner: &[u8; 32],
    uuid: &[u8; 16],
) -> Result<Option<(pl::ManifestV1, [u8; 32])>, String> {
    let adapter = app.state::<AppState>().freenet.clone();
    let ptr_ids = FreenetAdapter::contract_ids(PTR_WASM.to_vec(), &pl::make_ptr_params(owner, uuid));
    let pbytes = adapter.contract_get(&ptr_ids.instance, false).await.map_err(|e| {
        if e.to_string().contains("not found") {
            "no such shared playlist on the network (or not reachable yet)".to_string()
        } else {
            e.to_string()
        }
    })?;
    let p = pl::validate_pointer(owner, uuid, &pbytes).map_err(|e| format!("pointer failed verification: {e}"))?;
    let Some(r) = p.current.filter(|_| !p.withdrawn) else { return Ok(None) };
    let (m, _) = revision_by_hash(app, owner, &r.content_hash).await?;
    Ok(Some((m, r.content_hash)))
}

/// The local playlist behind a share was deleted, so the share goes with it:
/// the pointer is withdrawn while we can still reach it, and the ledger row
/// is dropped.
///
/// Dropping it is the point. The row is keyed by the LOCAL playlist id, which
/// SQLite hands to the next playlist created — and `publish` reuses the uuid
/// it finds under that key. Left behind, the next playlist would silently go
/// out as this one: same uuid, same proposal board, next revision, entirely
/// different tracks.
///
/// Best-effort on the network half. A withdraw that cannot be sent leaves the
/// last revision fetchable, which is true of every revision ever published —
/// nothing on Freenet is erasable, and withdrawing was never a delete.
pub(crate) async fn forget_playlist(app: &tauri::AppHandle, playlist_id: i64) {
    let key = playlist_id.to_string();
    let ledger = load(app);
    let rec = ledger.published.get(&key).cloned();
    // A follow points at the local copy by the same recyclable row id, so it
    // goes the same way — or the next playlist created would start taking
    // someone else's updates.
    let follows: Vec<String> = ledger
        .following
        .iter()
        .filter(|(_, f)| f.playlist_id == playlist_id)
        .map(|(k, _)| k.clone())
        .collect();
    if rec.is_none() && follows.is_empty() {
        return;
    }
    if let Some(rec) = rec.as_ref().filter(|r| !r.withdrawn) {
        // withdraw() takes this ledger's lock itself, so it cannot be called
        // while we hold it — the lock is not reentrant.
        if let Err(e) = withdraw(app, playlist_id).await {
            log::warn!(
                "shared playlists: deleting {:?} could not withdraw its pointer ({e}) — the last revision stays fetchable",
                rec.name
            );
        }
    }
    let _guard = ledger_lock::hold(LEDGER_FILE).await;
    let mut ledger = load(app);
    ledger.published.remove(&key);
    for k in &follows {
        ledger.following.remove(k);
    }
    match save(app, &ledger) {
        Ok(()) => {
            if let Some(rec) = &rec {
                log::info!("shared playlists: dropped the share record for deleted playlist {playlist_id} ({:?})", rec.name);
            }
            if !follows.is_empty() {
                log::info!("shared playlists: stopped following {} for deleted playlist {playlist_id}", follows.join(", "));
            }
        }
        // Worth a loud line: this is exactly the state where a new playlist
        // can inherit a dead uuid, or a dead follow.
        Err(e) => log::warn!("shared playlists: could not drop the records for playlist {playlist_id}: {e}"),
    }
}

/// Drop share records whose local playlist is no longer here. Fixing
/// `delete_playlist` only helps deletions from here on — a ledger that
/// already lost its playlist (an older build, or a row removed while the app
/// was down) still holds a key SQLite is free to hand out again. Runs once at
/// startup, before anything can create a playlist.
pub(crate) async fn reconcile_deleted(app: &tauri::AppHandle) {
    let lib = app.state::<LibraryState>();
    let l = lib.library.clone();
    let alive = match tauri::async_runtime::spawn_blocking(move || l.list_playlists()).await {
        Ok(Ok(rows)) => rows.into_iter().map(|p| p.id).collect::<BTreeSet<i64>>(),
        // Never guess here: an unreadable library would look like "every
        // playlist was deleted" and take every share record with it.
        other => {
            log::warn!("shared playlists: could not read the library to reconcile share records ({other:?})");
            return;
        }
    };
    let ledger = load(app);
    let orphans: BTreeSet<i64> = ledger
        .published
        .values()
        .map(|p| p.playlist_id)
        .chain(ledger.following.values().map(|f| f.playlist_id))
        .filter(|id| !alive.contains(id))
        .collect();
    for id in orphans {
        forget_playlist(app, id).await;
    }
}

/// The local playlist behind a published uuid, when this device published it.
pub(crate) fn published_for_uuid(app: &tauri::AppHandle, uuid_b58: &str) -> Option<Published> {
    load(app).published.into_values().find(|p| p.uuid == uuid_b58)
}

/// A verified shared playlist staged through the app's own import path.
pub async fn import(
    app: &tauri::AppHandle,
    lib: &State<'_, LibraryState>,
    owner_b58: &str,
    uuid_b58: &str,
) -> Result<ImportDraft, String> {
    let v = fetch(app, owner_b58, uuid_b58).await?;
    if v.withdrawn {
        return Err("the owner withdrew this playlist".into());
    }
    // Remembered for `follow`, which runs once the draft has become a local
    // playlist: the follow must start at the revision the copy was made
    // from, and a pointer read of its own could already name a newer one.
    staged().lock().unwrap().insert(
        follow_key(owner_b58, uuid_b58),
        Staged { hash: v.content_hash.clone(), seq: v.revision_seq, name: v.name.clone() },
    );
    let parsed = parsed_from(&v.name, &v.description, &v.entries);
    playlists::resolve_draft(lib, parsed, format!("SoulStream · {}", v.name), None).await
}

// ---------------------------------------------------------------- following

fn follow_key(owner_b58: &str, uuid_b58: &str) -> String {
    format!("{owner_b58}:{uuid_b58}")
}

/// What `import` fetched for an (owner, uuid), until `follow` claims it.
struct Staged {
    hash: String,
    seq: u64,
    name: String,
}
static STAGED: OnceLock<Mutex<HashMap<String, Staged>>> = OnceLock::new();
fn staged() -> &'static Mutex<HashMap<String, Staged>> {
    STAGED.get_or_init(Default::default)
}

/// Follow (owner, uuid) as local playlist `playlist_id`, from the revision
/// `import` fetched for it. With nothing staged (the app restarted between
/// import and commit) the pointer is read again; the only way that names a
/// different revision is an owner publish inside that window, and the first
/// Update then brings the copy in line.
pub async fn follow(
    app: &tauri::AppHandle,
    owner_b58: &str,
    uuid_b58: &str,
    playlist_id: i64,
) -> Result<Follow, String> {
    let key = follow_key(owner_b58, uuid_b58);
    let staged = staged().lock().unwrap().remove(&key);
    let (hash, seq, name) = match staged {
        Some(s) => (s.hash, s.seq, s.name),
        None => {
            let v = fetch(app, owner_b58, uuid_b58).await?;
            if v.withdrawn {
                return Err("the owner withdrew this playlist".into());
            }
            (v.content_hash, v.revision_seq, v.name)
        }
    };
    let now = now_secs();
    let rec = Follow {
        owner_key: owner_b58.to_string(),
        playlist_uuid: uuid_b58.to_string(),
        playlist_id,
        name,
        synced_hash: hash.clone(),
        synced_seq: seq,
        synced_at: now,
        followed_at: now,
        checked_at: now,
        seen_hash: hash,
        seen_seq: seq,
        withdrawn: false,
        added: vec![],
        removed: 0,
        last_error: None,
    };
    let _guard = ledger_lock::hold(LEDGER_FILE).await;
    let mut ledger = load(app);
    ledger.following.insert(key, rec.clone());
    save(app, &ledger)?;
    log::info!(
        "shared playlists: following {:?} as playlist {playlist_id} from revision {seq}",
        rec.name
    );
    Ok(rec)
}

pub async fn unfollow(app: &tauri::AppHandle, owner_b58: &str, uuid_b58: &str) -> Result<(), String> {
    let _guard = ledger_lock::hold(LEDGER_FILE).await;
    let mut ledger = load(app);
    let gone = ledger.following.remove(&follow_key(owner_b58, uuid_b58));
    let Some(gone) = gone else { return Err("not following that playlist".into()) };
    save(app, &ledger)?;
    log::info!("shared playlists: stopped following {:?}", gone.name);
    Ok(())
}

/// What the owner's current revision has that the synced one did not, in
/// the owner's order, and how many synced entries are gone. Occurrence ids
/// are the unit — the proposal board's rule: a track removed and added back
/// is a new entry, and a duplicate stays distinct.
fn diff_entries(synced: &[pl::EntryV1], current: &[pl::EntryV1]) -> (Vec<ShareEntry>, u32) {
    let had: BTreeSet<[u8; 16]> = synced.iter().map(|e| e.entry_id).collect();
    let has: BTreeSet<[u8; 16]> = current.iter().map(|e| e.entry_id).collect();
    let added = current.iter().filter(|e| !had.contains(&e.entry_id)).map(share_entry).collect();
    let removed = had.iter().filter(|id| !has.contains(*id)).count() as u32;
    (added, removed)
}

#[derive(Debug, Default, Serialize)]
pub struct FollowCheck {
    pub checked: u32,
    /// Follows whose pointer moved since the previous check.
    pub changed: u32,
    pub errors: u32,
}

/// What one check found: the pointer's revision, diffed against the synced
/// one. None when it still names the revision the previous check saw.
struct Seen {
    hash: String,
    seq: u64,
    withdrawn: bool,
    name: Option<String>,
    added: Vec<ShareEntry>,
    removed: u32,
}

async fn check_one(app: &tauri::AppHandle, f: &Follow) -> Result<Option<Seen>, String> {
    let owner = bytes32(&f.owner_key)?;
    let uuid = bytes16(&f.playlist_uuid)?;
    let adapter = app.state::<AppState>().freenet.clone();
    let ptr_ids = FreenetAdapter::contract_ids(PTR_WASM.to_vec(), &pl::make_ptr_params(&owner, &uuid));
    let pbytes = adapter.contract_get(&ptr_ids.instance, false).await.map_err(|e| e.to_string())?;
    let p = pl::validate_pointer(&owner, &uuid, &pbytes).map_err(|e| format!("pointer failed verification: {e}"))?;
    let Some(r) = p.current.filter(|_| !p.withdrawn) else {
        let already = f.withdrawn;
        return Ok((!already).then(|| Seen {
            hash: String::new(),
            seq: 0,
            withdrawn: true,
            name: None,
            added: vec![],
            removed: 0,
        }));
    };
    let hash = b58(&r.content_hash);
    if hash == f.seen_hash && !f.withdrawn {
        return Ok(None);
    }
    let (current, _) = revision_by_hash(app, &owner, &r.content_hash).await?;
    let (synced, _) = revision_by_hash(app, &owner, &bytes32(&f.synced_hash)?).await?;
    let (added, removed) = diff_entries(&synced.entries, &current.entries);
    Ok(Some(Seen {
        hash,
        seq: current.revision_seq,
        withdrawn: false,
        name: Some(current.name),
        added,
        removed,
    }))
}

/// One pass over every follow: read each pointer, and where it names a
/// revision the previous check did not, load it and diff it against the
/// synced one. The network reads run WITHOUT the ledger lock — a pointer
/// read is a ring search that can take seconds, and a share card arriving
/// meanwhile must not queue behind it — and each result is merged under
/// the lock afterwards, dropped if the follow moved on (an Update, an
/// unfollow) while the read was out.
pub async fn check_follows(app: &tauri::AppHandle) -> Result<FollowCheck, String> {
    let follows: Vec<Follow> = load(app).following.into_values().collect();
    let mut out = FollowCheck::default();
    for f in follows {
        out.checked += 1;
        let result = check_one(app, &f).await;
        let _guard = ledger_lock::hold(LEDGER_FILE).await;
        let mut ledger = load(app);
        let key = follow_key(&f.owner_key, &f.playlist_uuid);
        let Some(live) = ledger.following.get_mut(&key) else { continue };
        if live.synced_hash != f.synced_hash {
            continue;
        }
        live.checked_at = now_secs();
        match result {
            Ok(Some(seen)) => {
                out.changed += 1;
                if seen.withdrawn {
                    log::info!("shared playlists: {:?} was withdrawn by its owner", live.name);
                } else {
                    log::info!(
                        "shared playlists: {:?} is at revision {} — {} new, {} removed since revision {}",
                        live.name,
                        seen.seq,
                        seen.added.len(),
                        seen.removed,
                        live.synced_seq
                    );
                }
                live.seen_hash = seen.hash;
                live.seen_seq = seen.seq;
                live.withdrawn = seen.withdrawn;
                live.added = seen.added;
                live.removed = seen.removed;
                live.last_error = None;
                if let Some(n) = seen.name {
                    live.name = n;
                }
            }
            Ok(None) => live.last_error = None,
            Err(e) => {
                out.errors += 1;
                log::info!("shared playlists: could not check {:?}: {e}", live.name);
                live.last_error = Some(e);
            }
        }
        if let Err(e) = save(app, &ledger) {
            log::warn!("shared playlists: could not record the check of {:?}: {e}", f.name);
        }
    }
    Ok(out)
}

/// Bring the local copy up to the revision the last check saw: the new
/// entries resolve the way an import resolves them — the library first,
/// a Soulseek stream otherwise — and are appended in the owner's order.
/// Nothing is removed or reordered: the copy is the follower's to edit, and
/// a track the owner dropped is reported, never taken away. Held under the
/// ledger lock end to end (the resolve is local work), so a check landing
/// meanwhile merges against the new synced revision or not at all.
pub async fn follow_update(
    app: &tauri::AppHandle,
    lib: &State<'_, LibraryState>,
    owner_b58: &str,
    uuid_b58: &str,
) -> Result<u32, String> {
    let key = follow_key(owner_b58, uuid_b58);
    let _guard = ledger_lock::hold(LEDGER_FILE).await;
    let mut ledger = load(app);
    let f = ledger.following.get(&key).cloned().ok_or("not following that playlist")?;
    if f.withdrawn {
        return Err("the owner withdrew this playlist".into());
    }
    if f.seen_hash == f.synced_hash {
        return Ok(0);
    }
    let mut appended = 0;
    if !f.added.is_empty() {
        let parsed = parsed_from(&f.name, "", &f.added);
        let draft = playlists::resolve_draft(lib, parsed, format!("SoulStream · {}", f.name), None).await?;
        let rows: Vec<NewPlaylistItem> = draft.entries.into_iter().map(new_row).collect();
        let l = lib.library.clone();
        let pid = f.playlist_id;
        appended = tauri::async_runtime::spawn_blocking(move || l.add_playlist_items(pid, &rows, None, now_secs() as i64))
            .await
            .map_err(|e| e.to_string())?
            .map_err(|e| e.to_string())?;
        playlists::changed(app);
    }
    let live = ledger.following.get_mut(&key).ok_or("not following that playlist")?;
    live.synced_hash = f.seen_hash.clone();
    live.synced_seq = f.seen_seq;
    live.synced_at = now_secs();
    live.added.clear();
    live.removed = 0;
    save(app, &ledger)?;
    log::info!(
        "shared playlists: {:?} updated to revision {} — {appended} rows appended to playlist {}",
        f.name,
        f.seen_seq,
        f.playlist_id
    );
    Ok(appended)
}

// ---------------------------------------------------------------- share carry

pub async fn send(app: &tauri::AppHandle, username: &str, playlist_id: i64) -> Result<(), String> {
    let ledger = load(app);
    let rec = ledger
        .published
        .get(&playlist_id.to_string())
        .ok_or("publish the playlist first")?;
    if rec.withdrawn {
        return Err("that playlist is withdrawn — publish it again first".into());
    }
    let owner = own_key(app).await?;
    let r = ShareRef {
        owner,
        playlist_uuid: bytes16(&rec.uuid)?,
        name: rec.name.clone(),
        entry_count: rec.entry_count,
    };
    // Sealed through their inbox when they have one; the Soulseek carry otherwise.
    if let Some(link) = crate::slsk_link::known_link(app, username) {
        if let Ok(their) = bytes32(&link.persona_key) {
            let cargo = Cargo::Share(ShareCargo {
                owner,
                playlist_uuid: r.playlist_uuid,
                name: r.name.clone(),
                entry_count: r.entry_count,
            });
            match inbox::send(app, &their, &cargo).await {
                Ok(()) => {
                    log::info!("shared playlists: sent {:?} to {username} via inbox", rec.name);
                    return Ok(());
                }
                Err(e) => log::info!("shared playlists: inbox carry failed ({e}); trying the Soulseek carry"),
            }
        }
    }
    use base64::Engine;
    let bytes = bincode::serialize(&r).map_err(|e| e.to_string())?;
    let text = format!("{SHARE_PREFIX}{}", base64::engine::general_purpose::STANDARD.encode(bytes));
    send_pm(app, username, &text).await?;
    log::info!("shared playlists: sent {:?} to {username}", rec.name);
    Ok(())
}

async fn keep_card(app: &tauri::AppHandle, card: ShareCard) {
    let who = if card.from_username.is_empty() { card.from_key.clone() } else { card.from_username.clone() };
    let _guard = ledger_lock::hold(LEDGER_FILE).await;
    let mut ledger = load(app);
    ledger.inbox.retain(|c| c.id != card.id);
    ledger.inbox.push(card);
    if let Err(e) = save(app, &ledger) {
        log::warn!("shared playlists: could not keep the share from {who}: {e}");
    } else {
        log::info!("shared playlists: share received from {who}");
    }
}

/// Sealed cargo from inbox.rs: the envelope's signer is the sender.
pub(crate) async fn on_inbox_share(app: &tauri::AppHandle, sender: [u8; 32], s: ShareCargo) {
    let from_key = b58(&sender);
    let from_username = crate::slsk_link::username_for_key(app, &from_key).unwrap_or_default();
    keep_card(
        app,
        ShareCard {
            id: format!("{}:{}:{}", from_key, b58(&s.owner), b58(&s.playlist_uuid)),
            from_username,
            from_key,
            carry: "inbox".into(),
            owner_key: b58(&s.owner),
            playlist_uuid: b58(&s.playlist_uuid),
            name: cap(&s.name, pl::MAX_NAME_BYTES),
            entry_count: s.entry_count,
            received_at: now_secs(),
        },
    )
    .await;
}

/// The event pump hands private messages here after the link and friend
/// handshakes. True means it was a share reference, now a card in the Inbox.
pub fn intercept_pm(app: &tauri::AppHandle, from: &str, message: &str) -> bool {
    let Some(rest) = message.trim().strip_prefix(SHARE_PREFIX) else { return false };
    use base64::Engine;
    let Ok(bytes) = base64::engine::general_purpose::STANDARD.decode(rest.trim()) else { return true };
    let Ok(r) = bincode::deserialize::<ShareRef>(&bytes) else { return true };
    let card = ShareCard {
            id: format!("{}:{}:{}", normalize_username(from), b58(&r.owner), b58(&r.playlist_uuid)),
            from_username: from.to_string(),
            from_key: String::new(),
            carry: "pm".into(),
            owner_key: b58(&r.owner),
            playlist_uuid: b58(&r.playlist_uuid),
            name: cap(&r.name, pl::MAX_NAME_BYTES),
            entry_count: r.entry_count,
            received_at: now_secs(),
    };
    // Its own task: keeping a card now takes the ledger, and an inline await
    // here would block the event pump.
    let app = app.clone();
    tauri::async_runtime::spawn(async move { keep_card(&app, card).await });
    true
}

// ---------------------------------------------------------------- commands

#[tauri::command]
pub fn shared_playlists_mine(app: tauri::AppHandle) -> Vec<Published> {
    let mut v: Vec<_> = load(&app).published.into_values().collect();
    v.sort_by(|a, b| b.published_at.cmp(&a.published_at));
    v
}
#[tauri::command]
pub async fn shared_playlist_publish(
    app: tauri::AppHandle,
    lib: State<'_, LibraryState>,
    playlist_id: i64,
) -> Result<Published, String> {
    publish(&app, &lib, playlist_id).await
}
#[tauri::command]
pub async fn shared_playlist_withdraw(app: tauri::AppHandle, playlist_id: i64) -> Result<Published, String> {
    withdraw(&app, playlist_id).await
}
#[tauri::command]
pub async fn shared_playlist_fetch(app: tauri::AppHandle, owner: String, uuid: String) -> Result<ShareView, String> {
    fetch(&app, &owner, &uuid).await
}
#[tauri::command]
pub async fn shared_playlist_import(
    app: tauri::AppHandle,
    lib: State<'_, LibraryState>,
    owner: String,
    uuid: String,
) -> Result<ImportDraft, String> {
    import(&app, &lib, &owner, &uuid).await
}
#[tauri::command]
pub async fn share_send(app: tauri::AppHandle, username: String, playlist_id: i64) -> Result<(), String> {
    send(&app, &username, playlist_id).await
}
#[tauri::command]
pub fn shares_inbox(app: tauri::AppHandle) -> Vec<ShareCard> {
    let mut v = load(&app).inbox;
    v.sort_by(|a, b| b.received_at.cmp(&a.received_at));
    v
}
#[tauri::command]
pub async fn share_dismiss(app: tauri::AppHandle, id: String) -> Result<(), String> {
    let _guard = ledger_lock::hold(LEDGER_FILE).await;
    let mut ledger = load(&app);
    ledger.inbox.retain(|c| c.id != id);
    save(&app, &ledger)
}
#[tauri::command]
pub fn shared_playlists_following(app: tauri::AppHandle) -> Vec<Follow> {
    let mut v: Vec<_> = load(&app).following.into_values().collect();
    v.sort_by(|a, b| b.followed_at.cmp(&a.followed_at));
    v
}
#[tauri::command]
pub async fn shared_playlist_follow(
    app: tauri::AppHandle,
    owner: String,
    uuid: String,
    playlist_id: i64,
) -> Result<Follow, String> {
    follow(&app, &owner, &uuid, playlist_id).await
}
#[tauri::command]
pub async fn shared_playlist_unfollow(app: tauri::AppHandle, owner: String, uuid: String) -> Result<(), String> {
    unfollow(&app, &owner, &uuid).await
}
#[tauri::command]
pub async fn shared_playlists_check(app: tauri::AppHandle) -> Result<FollowCheck, String> {
    check_follows(&app).await
}
#[tauri::command]
pub async fn shared_playlist_follow_update(
    app: tauri::AppHandle,
    lib: State<'_, LibraryState>,
    owner: String,
    uuid: String,
) -> Result<u32, String> {
    follow_update(&app, &lib, &owner, &uuid).await
}

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

    fn entry(id: u8, position: u32, title: &str) -> pl::EntryV1 {
        pl::EntryV1 {
            entry_id: [id; 16],
            position,
            recording: pl::RecordingRef {
                musicbrainz_recording_id: None,
                isrcs: vec![],
                deezer_id: None,
                apple_music_id: None,
                title: title.into(),
                artists: vec!["Broadcast".into()],
                album: "Tender Buttons".into(),
                duration_ms: Some(200_000),
                release_year: Some(2005),
            },
            note: None,
            slsk_hint: None,
        }
    }

    /// The diff is by occurrence id, in the owner's order: a reorder is not
    /// news, a removal is counted, and a duplicate occurrence is its own row.
    #[test]
    fn a_follow_diff_is_by_occurrence_in_the_owners_order() {
        let synced = vec![entry(1, 0, "Black Cat"), entry(2, 1, "Tears in the Typing Pool"), entry(3, 2, "Corporeal")];
        let (added, removed) = diff_entries(&synced, &synced);
        assert!(added.is_empty());
        assert_eq!(removed, 0);

        // Reordered, one dropped, two appended, and Black Cat a second time
        // as a new occurrence.
        let current = vec![
            entry(3, 0, "Corporeal"),
            entry(1, 1, "Black Cat"),
            entry(9, 2, "Michael A Grammar"),
            entry(8, 3, "America's Boy"),
            entry(7, 4, "Black Cat"),
        ];
        let (added, removed) = diff_entries(&synced, &current);
        assert_eq!(
            added.iter().map(|e| e.title.as_str()).collect::<Vec<_>>(),
            ["Michael A Grammar", "America's Boy", "Black Cat"]
        );
        assert_eq!(added.iter().map(|e| e.position).collect::<Vec<_>>(), [2, 3, 4]);
        assert_eq!(removed, 1);
    }

    /// A ledger written before follows existed still loads, and a follow
    /// survives the round trip with its diff.
    #[test]
    fn the_ledger_reads_older_files_and_keeps_follows() {
        let old: Ledger = serde_json::from_str(r#"{"published":{},"inbox":[]}"#).unwrap();
        assert!(old.following.is_empty());

        let mut ledger = Ledger::default();
        let (added, removed) = diff_entries(&[entry(1, 0, "Black Cat")], &[entry(1, 0, "Black Cat"), entry(2, 1, "Corporeal")]);
        ledger.following.insert(
            follow_key("owner", "uuid"),
            Follow {
                owner_key: "owner".into(),
                playlist_uuid: "uuid".into(),
                playlist_id: 4,
                name: "Tender Buttons".into(),
                synced_hash: "h1".into(),
                synced_seq: 1,
                synced_at: 10,
                followed_at: 10,
                checked_at: 20,
                seen_hash: "h2".into(),
                seen_seq: 2,
                withdrawn: false,
                added,
                removed,
                last_error: None,
            },
        );
        let back: Ledger = serde_json::from_slice(&serde_json::to_vec(&ledger).unwrap()).unwrap();
        let f = &back.following["owner:uuid"];
        assert_eq!(f.playlist_id, 4);
        assert_eq!(f.added.len(), 1);
        assert_eq!(f.added[0].title, "Corporeal");
        assert_eq!(f.removed, 0);
    }

    /// The bundled contracts are commitments (copy-forward is disabled
    /// upstream): a changed wasm re-keys every published playlist.
    #[test]
    fn contract_generations_are_pinned() {
        let owner = [7u8; 32];
        let rev = FreenetAdapter::contract_ids(REV_WASM.to_vec(), &pl::make_rev_params(&owner, &[1u8; 32]));
        let ptr = FreenetAdapter::contract_ids(PTR_WASM.to_vec(), &pl::make_ptr_params(&owner, &[2u8; 16]));
        assert_eq!(rev.code_hash, "2JYdHHe3j59CVwCCKKeicjaF3rqoY2qjEbrYc95uUjkk");
        assert_eq!(ptr.code_hash, "Daqn9EMctrTniK5s5x4m9QBHYixsf265UMUAmLPATg5S");
    }
}