macltok

public

Star0

Watch0

50 commits 3 branches 0 tags 0 releases3.0 MiB

use futures_util::{StreamExt, stream};
use matrix_sdk::{
    Client,
    ruma::{
        api::{client::state::get_state_event_for_key, error::ErrorKind as ApiErrorKind},
        events::{GlobalAccountDataEventType, StateEventType, room::MediaSource},
        serde::{JsonObject, Raw},
    },
};
use ruma::{OwnedEventId, OwnedMxcUri, OwnedRoomId};
use serde_json::{Value as JsonValue, json};
use tokio::sync::mpsc;

use super::{
    commands::{self, Job},
    enqueue, media,
    mutations::{Mutation, MutationJob},
};
use crate::{
    CommandOutput, Error, ErrorKind, ImagePackSource, PackCommand, PackImageInfo, PackUsage, Patch,
    normalise::{ENABLED_ROOMS, ROOM_PACKS, USER_PACK, resolve_pack_id, stored_pack},
    transport::Responder,
};

/// read-modify-writes of one pack, which run in order per pack
pub(super) enum PackEdit {
    Create {
        display_name: String,
    },
    Delete,
    Meta {
        display_name: Patch<String>,
        avatar: Patch<OwnedMxcUri>,
        usage: Patch<Vec<PackUsage>>,
    },
    Put {
        shortcode: String,
        mxc: OwnedMxcUri,
        body: Option<String>,
        usage: Option<Vec<PackUsage>>,
        info: Option<PackImageInfo>,
    },
    Rename {
        from: String,
        to: String,
    },
    Remove {
        shortcode: String,
    },
    /// copies already on our homeserver. `create` is the pack meta for a room pack that doesn't exist yet
    Insert {
        images: Vec<(String, JsonObject)>,
        create: Option<JsonObject>,
    },
}

pub(super) async fn edit_pack(
    client: &Client,
    pack: &ImagePackSource,
    edit: PackEdit,
) -> Result<CommandOutput, Error> {
    let current = fetch_pack(client, pack).await?;
    let mut content = current.clone().unwrap_or_default();
    let mut output = CommandOutput::Done;
    match edit {
        PackEdit::Create { display_name } => {
            if !content.is_empty() {
                return Err(Error::new(
                    ErrorKind::Invalid,
                    "that state key already holds a pack",
                ));
            }
            content = object(json!({ "pack": { "display_name": display_name }, "images": {} }));
        }
        PackEdit::Delete => content.clear(),
        PackEdit::Meta {
            display_name,
            avatar,
            usage,
        } => {
            let meta = child(&mut content, "pack");
            patch(meta, "display_name", display_name, JsonValue::String);
            patch(meta, "avatar_url", avatar, |uri| uri.to_string().into());
            patch(meta, "usage", usage, usage_json);
        }
        PackEdit::Put {
            shortcode,
            mxc,
            body,
            usage,
            info,
        } => {
            let images = child(&mut content, "images");
            let mut image = match images.remove(&shortcode) {
                Some(JsonValue::Object(image)) => image,
                _ => JsonObject::new(),
            };
            let same_media = image.get("url").and_then(JsonValue::as_str) == Some(mxc.as_str());
            image.insert("url".to_owned(), mxc.to_string().into());
            if let Some(body) = body {
                image.insert("body".to_owned(), body.into());
            }
            if let Some(usage) = usage {
                image.insert("usage".to_owned(), usage_json(usage));
            }
            match info {
                Some(info) => {
                    image.insert("info".to_owned(), info_json(info));
                }
                None if !same_media => {
                    image.remove("info");
                }
                None => {}
            }
            images.insert(shortcode, image.into());
        }
        PackEdit::Rename { from, to } => {
            let images = child(&mut content, "images");
            if images.contains_key(&to) {
                return Err(Error::new(ErrorKind::Invalid, "that shortcode is taken"));
            }
            let image = images
                .remove(&from)
                .ok_or_else(|| Error::new(ErrorKind::Invalid, "no image has that shortcode"))?;
            images.insert(to, image);
        }
        PackEdit::Remove { shortcode } => {
            child(&mut content, "images").remove(&shortcode);
        }
        PackEdit::Insert { images, create } => {
            if content.is_empty() {
                match (create, pack) {
                    (Some(meta), _) => {
                        content.insert("pack".to_owned(), meta.into());
                    }
                    (None, ImagePackSource::Room { .. }) => {
                        return Err(Error::new(
                            ErrorKind::Invalid,
                            "there's no pack there to copy into",
                        ));
                    }
                    (None, ImagePackSource::User) => {}
                }
            }
            let target = child(&mut content, "images");
            let shortcodes = images
                .into_iter()
                .map(|(shortcode, image)| {
                    let free = free_shortcode(target, &shortcode);
                    target.insert(free.clone(), image.into());
                    free
                })
                .collect();
            output = CommandOutput::Cloned {
                shortcodes,
                skipped: Vec::new(),
            };
        }
    }
    if current.as_ref() != Some(&content) {
        write_pack(client, pack, content).await?;
    }
    Ok(output)
}

pub(super) async fn set_enabled_everywhere(
    client: &Client,
    room_id: OwnedRoomId,
    state_key: String,
    enabled: bool,
) -> Result<(), Error> {
    let account = client.account();
    let current = account
        .fetch_account_data(GlobalAccountDataEventType::from(ENABLED_ROOMS))
        .await
        .map_err(|source| Error::from_sdk("reading enabled packs failed", source))?
        .map(|raw| parse(raw.json().get()))
        .transpose()?;
    let mut content = current.clone().unwrap_or_default();
    let rooms = child(&mut content, "rooms");
    if enabled {
        child(rooms, room_id.as_str()).insert(state_key, JsonObject::new().into());
    } else {
        let state_keys = child(rooms, room_id.as_str());
        state_keys.remove(&state_key);
        if state_keys.is_empty() {
            rooms.remove(room_id.as_str());
        }
    }
    if current.as_ref() == Some(&content) {
        return Ok(());
    }
    account
        .set_account_data_raw(ENABLED_ROOMS.into(), raw(&content))
        .await
        .map(drop)
        .map_err(|source| Error::from_sdk("writing enabled packs failed", source))
}

/// the commands that download, upload or send, so they stay off the per-pack queue and only
/// take their turn for the final write
pub(super) async fn run_detached(
    client: Client,
    mutations: mpsc::Sender<MutationJob>,
    max_bytes: usize,
    command: PackCommand,
) -> Result<CommandOutput, Error> {
    match command {
        PackCommand::SendSticker {
            room_id,
            pack_id,
            shortcode,
        } => send_sticker(&client, room_id, &pack_id, &shortcode)
            .await
            .map(|()| CommandOutput::Done),
        PackCommand::Rehost(mxc) => rehost(&client, mxc, None, max_bytes)
            .await
            .map(CommandOutput::Mxc),
        PackCommand::CloneImage {
            from_pack,
            shortcode,
            to,
        } => {
            let image = pack_image(&client, &from_pack, &shortcode).await?;
            let image = rehost_image(&client, image, max_bytes).await?;
            insert(&mutations, to, vec![(shortcode, image)], None, Vec::new()).await
        }
        PackCommand::CloneSticker {
            room_id,
            event_id,
            to,
            shortcode,
        } => {
            let image = sticker_image(&client, room_id, event_id).await?;
            let image = rehost_image(&client, image, max_bytes).await?;
            insert(&mutations, to, vec![(shortcode, image)], None, Vec::new()).await
        }
        PackCommand::ClonePack {
            from_pack,
            to,
            display_name,
        } => clone_pack(&client, &mutations, max_bytes, &from_pack, to, display_name).await,
        _ => Err(Error::new(ErrorKind::Other, "pack command was misrouted")),
    }
}

async fn clone_pack(
    client: &Client,
    mutations: &mpsc::Sender<MutationJob>,
    max_bytes: usize,
    from_pack: &str,
    to: ImagePackSource,
    display_name: Option<String>,
) -> Result<CommandOutput, Error> {
    let source = resolve_pack_id(client, from_pack)
        .ok_or_else(|| Error::new(ErrorKind::Invalid, "no pack has that id"))?;
    let content = stored_pack(client, &source)
        .await?
        .ok_or_else(|| Error::new(ErrorKind::Invalid, "that pack is empty"))?;
    let mut meta = match content.get("pack") {
        Some(JsonValue::Object(meta)) => meta.clone(),
        _ => JsonObject::new(),
    };
    let images: Vec<(String, JsonObject)> = match content.get("images") {
        Some(JsonValue::Object(images)) => images
            .iter()
            .filter_map(|(shortcode, image)| {
                let image = with_pack_usage(image.as_object()?.clone(), &meta);
                Some((shortcode.clone(), image))
            })
            .collect(),
        _ => Vec::new(),
    };
    let results: Vec<_> = stream::iter(images)
        .map(|(shortcode, image)| async move {
            match rehost_image(client, image, max_bytes).await {
                Ok(image) => Ok((shortcode, image)),
                Err(error) => {
                    tracing::debug!(%error, shortcode, "skipping an image that wouldn't rehost");
                    Err(shortcode)
                }
            }
        })
        .buffered(4)
        .collect()
        .await;
    let (copies, skipped): (Vec<_>, Vec<_>) = results.into_iter().partition(Result::is_ok);
    let copies = copies.into_iter().filter_map(Result::ok).collect();
    let skipped = skipped.into_iter().filter_map(Result::err).collect();
    let create = match to {
        ImagePackSource::User => None,
        ImagePackSource::Room { .. } => {
            if let Some(display_name) = display_name {
                meta.insert("display_name".to_owned(), display_name.into());
            }
            meta.remove("usage");
            if let Some(avatar) = meta.remove("avatar_url").and_then(|uri| mxc(&uri))
                && let Ok(avatar) = rehost(client, avatar, None, max_bytes).await
            {
                meta.insert("avatar_url".to_owned(), avatar.to_string().into());
            }
            Some(meta)
        }
    };
    insert(mutations, to, copies, create, skipped).await
}

async fn insert(
    mutations: &mpsc::Sender<MutationJob>,
    pack: ImagePackSource,
    images: Vec<(String, JsonObject)>,
    create: Option<JsonObject>,
    skipped: Vec<String>,
) -> Result<CommandOutput, Error> {
    let (reply, answer) = Responder::channel();
    let command = Mutation::Pack {
        pack,
        edit: PackEdit::Insert { images, create },
    };
    enqueue(mutations, Job { command, reply });
    match answer.await? {
        CommandOutput::Cloned { shortcodes, .. } => Ok(CommandOutput::Cloned {
            shortcodes,
            skipped,
        }),
        output => Ok(output),
    }
}

async fn send_sticker(
    client: &Client,
    room_id: OwnedRoomId,
    pack_id: &str,
    shortcode: &str,
) -> Result<(), Error> {
    let room = client.get_room(&room_id).ok_or_else(Error::room_unknown)?;
    let image = pack_image(client, pack_id, shortcode).await?;
    let url = image
        .get("url")
        .and_then(mxc)
        .ok_or_else(|| Error::new(ErrorKind::Invalid, "that image has no usable url"))?;
    let content = json!({
        "body": image.get("body").and_then(JsonValue::as_str).unwrap_or(shortcode),
        "url": url,
        "info": image.get("info").cloned().unwrap_or_else(|| json!({})),
    });
    room.send_queue()
        .send_raw(raw(&object(content)), "m.sticker".to_owned())
        .await
        .map(drop)
        .map_err(|source| {
            Error::with_source(ErrorKind::Store, "queueing the sticker failed", source)
        })
}

/// the image with its pack's usage folded in, so it means the same once it leaves the pack
async fn pack_image(client: &Client, pack_id: &str, shortcode: &str) -> Result<JsonObject, Error> {
    let source = resolve_pack_id(client, pack_id)
        .ok_or_else(|| Error::new(ErrorKind::Invalid, "no pack has that id"))?;
    let content = stored_pack(client, &source).await?.unwrap_or_default();
    let image = content
        .get("images")
        .and_then(|images| images.get(shortcode))
        .and_then(JsonValue::as_object)
        .ok_or_else(|| Error::new(ErrorKind::Invalid, "no image has that shortcode"))?
        .clone();
    let meta = content
        .get("pack")
        .and_then(JsonValue::as_object)
        .cloned()
        .unwrap_or_default();
    Ok(with_pack_usage(image, &meta))
}

async fn sticker_image(
    client: &Client,
    room_id: OwnedRoomId,
    event_id: OwnedEventId,
) -> Result<JsonObject, Error> {
    let room = client.get_room(&room_id).ok_or_else(Error::room_unknown)?;
    let event = room
        .load_or_fetch_event(&event_id, None)
        .await
        .map_err(|source| Error::from_sdk("loading the sticker failed", source))?;
    let raw = event.raw();
    if raw.get_field::<String>("type").ok().flatten().as_deref() != Some("m.sticker") {
        return Err(Error::new(ErrorKind::Invalid, "that event isn't a sticker"));
    }
    let content = raw
        .get_field::<JsonObject>("content")
        .ok()
        .flatten()
        .unwrap_or_default();
    let mut image = JsonObject::new();
    for key in ["url", "body", "info"] {
        if let Some(value) = content.get(key) {
            image.insert(key.to_owned(), value.clone());
        }
    }
    image.insert("usage".to_owned(), usage_json(vec![PackUsage::Sticker]));
    Ok(image)
}

async fn rehost_image(
    client: &Client,
    mut image: JsonObject,
    max_bytes: usize,
) -> Result<JsonObject, Error> {
    let url = image
        .get("url")
        .and_then(mxc)
        .ok_or_else(|| Error::new(ErrorKind::Invalid, "that image has no usable url"))?;
    let mimetype = image
        .get("info")
        .and_then(|info| info.get("mimetype"))
        .and_then(JsonValue::as_str)
        .map(str::to_owned);
    let url = rehost(client, url, mimetype, max_bytes).await?;
    image.insert("url".to_owned(), url.to_string().into());
    Ok(image)
}

pub(super) async fn rehost(
    client: &Client,
    mxc: OwnedMxcUri,
    mimetype: Option<String>,
    max_bytes: usize,
) -> Result<OwnedMxcUri, Error> {
    let homeserver = client.user_id().map(ruma::UserId::server_name);
    if mxc
        .server_name()
        .ok()
        .is_some_and(|server| Some(server) == homeserver)
    {
        return Ok(mxc);
    }
    let media = media::download_file(client, MediaSource::Plain(mxc), mimetype, max_bytes).await?;
    commands::upload_media(client, &media.mimetype, media.bytes.to_vec(), max_bytes).await
}

/// the server's copy, since the store lags a sync behind and an edit made just before would be lost
async fn fetch_pack(client: &Client, pack: &ImagePackSource) -> Result<Option<JsonObject>, Error> {
    match pack {
        ImagePackSource::User => client
            .account()
            .fetch_account_data(GlobalAccountDataEventType::from(USER_PACK))
            .await
            .map_err(|source| Error::from_sdk("reading the pack failed", source))?
            .map(|raw| parse(raw.json().get()))
            .transpose(),
        ImagePackSource::Room { room_id, state_key } => {
            let request = get_state_event_for_key::v3::Request::new(
                room_id.clone(),
                StateEventType::from(ROOM_PACKS),
                state_key.clone(),
            );
            match client.send(request).await {
                Ok(response) => parse(response.event_or_content.get()).map(Some),
                Err(error) if error.client_api_error_kind() == Some(&ApiErrorKind::NotFound) => {
                    Ok(None)
                }
                Err(source) => Err(Error::from_http("reading the pack failed", source)),
            }
        }
    }
}

async fn write_pack(
    client: &Client,
    pack: &ImagePackSource,
    content: JsonObject,
) -> Result<(), Error> {
    let written = match pack {
        ImagePackSource::User => client
            .account()
            .set_account_data_raw(USER_PACK.into(), raw(&content))
            .await
            .map(drop),
        ImagePackSource::Room { room_id, state_key } => client
            .get_room(room_id)
            .ok_or_else(Error::room_unknown)?
            .send_state_event_raw(ROOM_PACKS, state_key, JsonValue::Object(content))
            .await
            .map(drop),
    };
    written.map_err(|source| Error::from_sdk("writing the pack failed", source))
}

fn with_pack_usage(mut image: JsonObject, meta: &JsonObject) -> JsonObject {
    if !image.contains_key("usage")
        && let Some(usage) = meta.get("usage")
    {
        image.insert("usage".to_owned(), usage.clone());
    }
    image
}

fn free_shortcode(images: &JsonObject, wanted: &str) -> String {
    if !images.contains_key(wanted) {
        return wanted.to_owned();
    }
    (2..)
        .map(|n| format!("{wanted}-{n}"))
        .find(|candidate| !images.contains_key(candidate))
        .expect("an unbounded range finds a free name")
}

/// the object at `key`, made or replaced if it was missing or malformed
fn child<'a>(parent: &'a mut JsonObject, key: &str) -> &'a mut JsonObject {
    let slot = parent
        .entry(key.to_owned())
        .or_insert_with(|| JsonObject::new().into());
    if !slot.is_object() {
        *slot = JsonObject::new().into();
    }
    slot.as_object_mut().expect("just made it an object")
}

fn patch<T>(
    meta: &mut JsonObject,
    key: &str,
    change: Patch<T>,
    to_json: impl FnOnce(T) -> JsonValue,
) {
    match change {
        Patch::Keep => {}
        Patch::Clear => {
            meta.remove(key);
        }
        Patch::Set(value) => {
            meta.insert(key.to_owned(), to_json(value));
        }
    }
}

fn usage_json(usage: Vec<PackUsage>) -> JsonValue {
    usage
        .into_iter()
        .map(|usage| match usage {
            PackUsage::Emoticon => "emoticon",
            PackUsage::Sticker => "sticker",
        })
        .collect::<Vec<_>>()
        .into()
}

fn info_json(info: PackImageInfo) -> JsonValue {
    let mut out = JsonObject::new();
    if let Some(mimetype) = info.mimetype {
        out.insert("mimetype".to_owned(), mimetype.into());
    }
    for (key, value) in [
        ("size", info.size_bytes),
        ("w", info.width),
        ("h", info.height),
    ] {
        if let Some(value) = value {
            out.insert(key.to_owned(), value.into());
        }
    }
    out.into()
}

fn mxc(value: &JsonValue) -> Option<OwnedMxcUri> {
    let uri = OwnedMxcUri::from(value.as_str()?);
    uri.is_valid().then_some(uri)
}

fn object(value: JsonValue) -> JsonObject {
    match value {
        JsonValue::Object(object) => object,
        _ => JsonObject::new(),
    }
}

fn parse(json: &str) -> Result<JsonObject, Error> {
    serde_json::from_str(json)
        .map_err(|source| Error::with_source(ErrorKind::Server, "the pack is malformed", source))
}

fn raw<T>(content: &JsonObject) -> Raw<T> {
    Raw::new(content)
        .expect("a json object always serialises")
        .cast_unchecked()
}

1	use futures_util::{StreamExt, stream};
2	use matrix_sdk::{
3	Client,
4	ruma::{
5	api::{client::state::get_state_event_for_key, error::ErrorKind as ApiErrorKind},
6	events::{GlobalAccountDataEventType, StateEventType, room::MediaSource},
7	serde::{JsonObject, Raw},
8	},
9	};
10	use ruma::{OwnedEventId, OwnedMxcUri, OwnedRoomId};
11	use serde_json::{Value as JsonValue, json};
12	use tokio::sync::mpsc;
13
14	use super::{
15	commands::{self, Job},
16	enqueue, media,
17	mutations::{Mutation, MutationJob},
18	};
19	use crate::{
20	CommandOutput, Error, ErrorKind, ImagePackSource, PackCommand, PackImageInfo, PackUsage, Patch,
21	normalise::{ENABLED_ROOMS, ROOM_PACKS, USER_PACK, resolve_pack_id, stored_pack},
22	transport::Responder,
23	};
24
25	/// read-modify-writes of one pack, which run in order per pack
26	pub(super) enum PackEdit {
27	Create {
28	display_name: String,
29	},
30	Delete,
31	Meta {
32	display_name: Patch<String>,
33	avatar: Patch<OwnedMxcUri>,
34	usage: Patch<Vec<PackUsage>>,
35	},
36	Put {
37	shortcode: String,
38	mxc: OwnedMxcUri,
39	body: Option<String>,
40	usage: Option<Vec<PackUsage>>,
41	info: Option<PackImageInfo>,
42	},
43	Rename {
44	from: String,
45	to: String,
46	},
47	Remove {
48	shortcode: String,
49	},
50	/// copies already on our homeserver. `create` is the pack meta for a room pack that doesn't exist yet
51	Insert {
52	images: Vec<(String, JsonObject)>,
53	create: Option<JsonObject>,
54	},
55	}
56
57	pub(super) async fn edit_pack(
58	client: &Client,
59	pack: &ImagePackSource,
60	edit: PackEdit,
61	) -> Result<CommandOutput, Error> {
62	let current = fetch_pack(client, pack).await?;
63	let mut content = current.clone().unwrap_or_default();
64	let mut output = CommandOutput::Done;
65	match edit {
66	PackEdit::Create { display_name } => {
67	if !content.is_empty() {
68	return Err(Error::new(
69	ErrorKind::Invalid,
70	"that state key already holds a pack",
71	));
72	}
73	content = object(json!({ "pack": { "display_name": display_name }, "images": {} }));
74	}
75	PackEdit::Delete => content.clear(),
76	PackEdit::Meta {
77	display_name,
78	avatar,
79	usage,
80	} => {
81	let meta = child(&mut content, "pack");
82	patch(meta, "display_name", display_name, JsonValue::String);
83	patch(meta, "avatar_url", avatar, \|uri\| uri.to_string().into());
84	patch(meta, "usage", usage, usage_json);
85	}
86	PackEdit::Put {
87	shortcode,
88	mxc,
89	body,
90	usage,
91	info,
92	} => {
93	let images = child(&mut content, "images");
94	let mut image = match images.remove(&shortcode) {
95	Some(JsonValue::Object(image)) => image,
96	_ => JsonObject::new(),
97	};
98	let same_media = image.get("url").and_then(JsonValue::as_str) == Some(mxc.as_str());
99	image.insert("url".to_owned(), mxc.to_string().into());
100	if let Some(body) = body {
101	image.insert("body".to_owned(), body.into());
102	}
103	if let Some(usage) = usage {
104	image.insert("usage".to_owned(), usage_json(usage));
105	}
106	match info {
107	Some(info) => {
108	image.insert("info".to_owned(), info_json(info));
109	}
110	None if !same_media => {
111	image.remove("info");
112	}
113	None => {}
114	}
115	images.insert(shortcode, image.into());
116	}
117	PackEdit::Rename { from, to } => {
118	let images = child(&mut content, "images");
119	if images.contains_key(&to) {
120	return Err(Error::new(ErrorKind::Invalid, "that shortcode is taken"));
121	}
122	let image = images
123	.remove(&from)
124	.ok_or_else(\|\| Error::new(ErrorKind::Invalid, "no image has that shortcode"))?;
125	images.insert(to, image);
126	}
127	PackEdit::Remove { shortcode } => {
128	child(&mut content, "images").remove(&shortcode);
129	}
130	PackEdit::Insert { images, create } => {
131	if content.is_empty() {
132	match (create, pack) {
133	(Some(meta), _) => {
134	content.insert("pack".to_owned(), meta.into());
135	}
136	(None, ImagePackSource::Room { .. }) => {
137	return Err(Error::new(
138	ErrorKind::Invalid,
139	"there's no pack there to copy into",
140	));
141	}
142	(None, ImagePackSource::User) => {}
143	}
144	}
145	let target = child(&mut content, "images");
146	let shortcodes = images
147	.into_iter()
148	.map(\|(shortcode, image)\| {
149	let free = free_shortcode(target, &shortcode);
150	target.insert(free.clone(), image.into());
151	free
152	})
153	.collect();
154	output = CommandOutput::Cloned {
155	shortcodes,
156	skipped: Vec::new(),
157	};
158	}
159	}
160	if current.as_ref() != Some(&content) {
161	write_pack(client, pack, content).await?;
162	}
163	Ok(output)
164	}
165
166	pub(super) async fn set_enabled_everywhere(
167	client: &Client,
168	room_id: OwnedRoomId,
169	state_key: String,
170	enabled: bool,
171	) -> Result<(), Error> {
172	let account = client.account();
173	let current = account
174	.fetch_account_data(GlobalAccountDataEventType::from(ENABLED_ROOMS))
175	.await
176	.map_err(\|source\| Error::from_sdk("reading enabled packs failed", source))?
177	.map(\|raw\| parse(raw.json().get()))
178	.transpose()?;
179	let mut content = current.clone().unwrap_or_default();
180	let rooms = child(&mut content, "rooms");
181	if enabled {
182	child(rooms, room_id.as_str()).insert(state_key, JsonObject::new().into());
183	} else {
184	let state_keys = child(rooms, room_id.as_str());
185	state_keys.remove(&state_key);
186	if state_keys.is_empty() {
187	rooms.remove(room_id.as_str());
188	}
189	}
190	if current.as_ref() == Some(&content) {
191	return Ok(());
192	}
193	account
194	.set_account_data_raw(ENABLED_ROOMS.into(), raw(&content))
195	.await
196	.map(drop)
197	.map_err(\|source\| Error::from_sdk("writing enabled packs failed", source))
198	}
199
200	/// the commands that download, upload or send, so they stay off the per-pack queue and only
201	/// take their turn for the final write
202	pub(super) async fn run_detached(
203	client: Client,
204	mutations: mpsc::Sender<MutationJob>,
205	max_bytes: usize,
206	command: PackCommand,
207	) -> Result<CommandOutput, Error> {
208	match command {
209	PackCommand::SendSticker {
210	room_id,
211	pack_id,
212	shortcode,
213	} => send_sticker(&client, room_id, &pack_id, &shortcode)
214	.await
215	.map(\|()\| CommandOutput::Done),
216	PackCommand::Rehost(mxc) => rehost(&client, mxc, None, max_bytes)
217	.await
218	.map(CommandOutput::Mxc),
219	PackCommand::CloneImage {
220	from_pack,
221	shortcode,
222	to,
223	} => {
224	let image = pack_image(&client, &from_pack, &shortcode).await?;
225	let image = rehost_image(&client, image, max_bytes).await?;
226	insert(&mutations, to, vec![(shortcode, image)], None, Vec::new()).await
227	}
228	PackCommand::CloneSticker {
229	room_id,
230	event_id,
231	to,
232	shortcode,
233	} => {
234	let image = sticker_image(&client, room_id, event_id).await?;
235	let image = rehost_image(&client, image, max_bytes).await?;
236	insert(&mutations, to, vec![(shortcode, image)], None, Vec::new()).await
237	}
238	PackCommand::ClonePack {
239	from_pack,
240	to,
241	display_name,
242	} => clone_pack(&client, &mutations, max_bytes, &from_pack, to, display_name).await,
243	_ => Err(Error::new(ErrorKind::Other, "pack command was misrouted")),
244	}
245	}
246
247	async fn clone_pack(
248	client: &Client,
249	mutations: &mpsc::Sender<MutationJob>,
250	max_bytes: usize,
251	from_pack: &str,
252	to: ImagePackSource,
253	display_name: Option<String>,
254	) -> Result<CommandOutput, Error> {
255	let source = resolve_pack_id(client, from_pack)
256	.ok_or_else(\|\| Error::new(ErrorKind::Invalid, "no pack has that id"))?;
257	let content = stored_pack(client, &source)
258	.await?
259	.ok_or_else(\|\| Error::new(ErrorKind::Invalid, "that pack is empty"))?;
260	let mut meta = match content.get("pack") {
261	Some(JsonValue::Object(meta)) => meta.clone(),
262	_ => JsonObject::new(),
263	};
264	let images: Vec<(String, JsonObject)> = match content.get("images") {
265	Some(JsonValue::Object(images)) => images
266	.iter()
267	.filter_map(\|(shortcode, image)\| {
268	let image = with_pack_usage(image.as_object()?.clone(), &meta);
269	Some((shortcode.clone(), image))
270	})
271	.collect(),
272	_ => Vec::new(),
273	};
274	let results: Vec<_> = stream::iter(images)
275	.map(\|(shortcode, image)\| async move {
276	match rehost_image(client, image, max_bytes).await {
277	Ok(image) => Ok((shortcode, image)),
278	Err(error) => {
279	tracing::debug!(%error, shortcode, "skipping an image that wouldn't rehost");
280	Err(shortcode)
281	}
282	}
283	})
284	.buffered(4)
285	.collect()
286	.await;
287	let (copies, skipped): (Vec<_>, Vec<_>) = results.into_iter().partition(Result::is_ok);
288	let copies = copies.into_iter().filter_map(Result::ok).collect();
289	let skipped = skipped.into_iter().filter_map(Result::err).collect();
290	let create = match to {
291	ImagePackSource::User => None,
292	ImagePackSource::Room { .. } => {
293	if let Some(display_name) = display_name {
294	meta.insert("display_name".to_owned(), display_name.into());
295	}
296	meta.remove("usage");
297	if let Some(avatar) = meta.remove("avatar_url").and_then(\|uri\| mxc(&uri))
298	&& let Ok(avatar) = rehost(client, avatar, None, max_bytes).await
299	{
300	meta.insert("avatar_url".to_owned(), avatar.to_string().into());
301	}
302	Some(meta)
303	}
304	};
305	insert(mutations, to, copies, create, skipped).await
306	}
307
308	async fn insert(
309	mutations: &mpsc::Sender<MutationJob>,
310	pack: ImagePackSource,
311	images: Vec<(String, JsonObject)>,
312	create: Option<JsonObject>,
313	skipped: Vec<String>,
314	) -> Result<CommandOutput, Error> {
315	let (reply, answer) = Responder::channel();
316	let command = Mutation::Pack {
317	pack,
318	edit: PackEdit::Insert { images, create },
319	};
320	enqueue(mutations, Job { command, reply });
321	match answer.await? {
322	CommandOutput::Cloned { shortcodes, .. } => Ok(CommandOutput::Cloned {
323	shortcodes,
324	skipped,
325	}),
326	output => Ok(output),
327	}
328	}
329
330	async fn send_sticker(
331	client: &Client,
332	room_id: OwnedRoomId,
333	pack_id: &str,
334	shortcode: &str,
335	) -> Result<(), Error> {
336	let room = client.get_room(&room_id).ok_or_else(Error::room_unknown)?;
337	let image = pack_image(client, pack_id, shortcode).await?;
338	let url = image
339	.get("url")
340	.and_then(mxc)
341	.ok_or_else(\|\| Error::new(ErrorKind::Invalid, "that image has no usable url"))?;
342	let content = json!({
343	"body": image.get("body").and_then(JsonValue::as_str).unwrap_or(shortcode),
344	"url": url,
345	"info": image.get("info").cloned().unwrap_or_else(\|\| json!({})),
346	});
347	room.send_queue()
348	.send_raw(raw(&object(content)), "m.sticker".to_owned())
349	.await
350	.map(drop)
351	.map_err(\|source\| {
352	Error::with_source(ErrorKind::Store, "queueing the sticker failed", source)
353	})
354	}
355
356	/// the image with its pack's usage folded in, so it means the same once it leaves the pack
357	async fn pack_image(client: &Client, pack_id: &str, shortcode: &str) -> Result<JsonObject, Error> {
358	let source = resolve_pack_id(client, pack_id)
359	.ok_or_else(\|\| Error::new(ErrorKind::Invalid, "no pack has that id"))?;
360	let content = stored_pack(client, &source).await?.unwrap_or_default();
361	let image = content
362	.get("images")
363	.and_then(\|images\| images.get(shortcode))
364	.and_then(JsonValue::as_object)
365	.ok_or_else(\|\| Error::new(ErrorKind::Invalid, "no image has that shortcode"))?
366	.clone();
367	let meta = content
368	.get("pack")
369	.and_then(JsonValue::as_object)
370	.cloned()
371	.unwrap_or_default();
372	Ok(with_pack_usage(image, &meta))
373	}
374
375	async fn sticker_image(
376	client: &Client,
377	room_id: OwnedRoomId,
378	event_id: OwnedEventId,
379	) -> Result<JsonObject, Error> {
380	let room = client.get_room(&room_id).ok_or_else(Error::room_unknown)?;
381	let event = room
382	.load_or_fetch_event(&event_id, None)
383	.await
384	.map_err(\|source\| Error::from_sdk("loading the sticker failed", source))?;
385	let raw = event.raw();
386	if raw.get_field::<String>("type").ok().flatten().as_deref() != Some("m.sticker") {
387	return Err(Error::new(ErrorKind::Invalid, "that event isn't a sticker"));
388	}
389	let content = raw
390	.get_field::<JsonObject>("content")
391	.ok()
392	.flatten()
393	.unwrap_or_default();
394	let mut image = JsonObject::new();
395	for key in ["url", "body", "info"] {
396	if let Some(value) = content.get(key) {
397	image.insert(key.to_owned(), value.clone());
398	}
399	}
400	image.insert("usage".to_owned(), usage_json(vec![PackUsage::Sticker]));
401	Ok(image)
402	}
403
404	async fn rehost_image(
405	client: &Client,
406	mut image: JsonObject,
407	max_bytes: usize,
408	) -> Result<JsonObject, Error> {
409	let url = image
410	.get("url")
411	.and_then(mxc)
412	.ok_or_else(\|\| Error::new(ErrorKind::Invalid, "that image has no usable url"))?;
413	let mimetype = image
414	.get("info")
415	.and_then(\|info\| info.get("mimetype"))
416	.and_then(JsonValue::as_str)
417	.map(str::to_owned);
418	let url = rehost(client, url, mimetype, max_bytes).await?;
419	image.insert("url".to_owned(), url.to_string().into());
420	Ok(image)
421	}
422
423	pub(super) async fn rehost(
424	client: &Client,
425	mxc: OwnedMxcUri,
426	mimetype: Option<String>,
427	max_bytes: usize,
428	) -> Result<OwnedMxcUri, Error> {
429	let homeserver = client.user_id().map(ruma::UserId::server_name);
430	if mxc
431	.server_name()
432	.ok()
433	.is_some_and(\|server\| Some(server) == homeserver)
434	{
435	return Ok(mxc);
436	}
437	let media = media::download_file(client, MediaSource::Plain(mxc), mimetype, max_bytes).await?;
438	commands::upload_media(client, &media.mimetype, media.bytes.to_vec(), max_bytes).await
439	}
440
441	/// the server's copy, since the store lags a sync behind and an edit made just before would be lost
442	async fn fetch_pack(client: &Client, pack: &ImagePackSource) -> Result<Option<JsonObject>, Error> {
443	match pack {
444	ImagePackSource::User => client
445	.account()
446	.fetch_account_data(GlobalAccountDataEventType::from(USER_PACK))
447	.await
448	.map_err(\|source\| Error::from_sdk("reading the pack failed", source))?
449	.map(\|raw\| parse(raw.json().get()))
450	.transpose(),
451	ImagePackSource::Room { room_id, state_key } => {
452	let request = get_state_event_for_key::v3::Request::new(
453	room_id.clone(),
454	StateEventType::from(ROOM_PACKS),
455	state_key.clone(),
456	);
457	match client.send(request).await {
458	Ok(response) => parse(response.event_or_content.get()).map(Some),
459	Err(error) if error.client_api_error_kind() == Some(&ApiErrorKind::NotFound) => {
460	Ok(None)
461	}
462	Err(source) => Err(Error::from_http("reading the pack failed", source)),
463	}
464	}
465	}
466	}
467
468	async fn write_pack(
469	client: &Client,
470	pack: &ImagePackSource,
471	content: JsonObject,
472	) -> Result<(), Error> {
473	let written = match pack {
474	ImagePackSource::User => client
475	.account()
476	.set_account_data_raw(USER_PACK.into(), raw(&content))
477	.await
478	.map(drop),
479	ImagePackSource::Room { room_id, state_key } => client
480	.get_room(room_id)
481	.ok_or_else(Error::room_unknown)?
482	.send_state_event_raw(ROOM_PACKS, state_key, JsonValue::Object(content))
483	.await
484	.map(drop),
485	};
486	written.map_err(\|source\| Error::from_sdk("writing the pack failed", source))
487	}
488
489	fn with_pack_usage(mut image: JsonObject, meta: &JsonObject) -> JsonObject {
490	if !image.contains_key("usage")
491	&& let Some(usage) = meta.get("usage")
492	{
493	image.insert("usage".to_owned(), usage.clone());
494	}
495	image
496	}
497
498	fn free_shortcode(images: &JsonObject, wanted: &str) -> String {
499	if !images.contains_key(wanted) {
500	return wanted.to_owned();
501	}
502	(2..)
503	.map(\|n\| format!("{wanted}-{n}"))
504	.find(\|candidate\| !images.contains_key(candidate))
505	.expect("an unbounded range finds a free name")
506	}
507
508	/// the object at `key`, made or replaced if it was missing or malformed
509	fn child<'a>(parent: &'a mut JsonObject, key: &str) -> &'a mut JsonObject {
510	let slot = parent
511	.entry(key.to_owned())
512	.or_insert_with(\|\| JsonObject::new().into());
513	if !slot.is_object() {
514	*slot = JsonObject::new().into();
515	}
516	slot.as_object_mut().expect("just made it an object")
517	}
518
519	fn patch<T>(
520	meta: &mut JsonObject,
521	key: &str,
522	change: Patch<T>,
523	to_json: impl FnOnce(T) -> JsonValue,
524	) {
525	match change {
526	Patch::Keep => {}
527	Patch::Clear => {
528	meta.remove(key);
529	}
530	Patch::Set(value) => {
531	meta.insert(key.to_owned(), to_json(value));
532	}
533	}
534	}
535
536	fn usage_json(usage: Vec<PackUsage>) -> JsonValue {
537	usage
538	.into_iter()
539	.map(\|usage\| match usage {
540	PackUsage::Emoticon => "emoticon",
541	PackUsage::Sticker => "sticker",
542	})
543	.collect::<Vec<_>>()
544	.into()
545	}
546
547	fn info_json(info: PackImageInfo) -> JsonValue {
548	let mut out = JsonObject::new();
549	if let Some(mimetype) = info.mimetype {
550	out.insert("mimetype".to_owned(), mimetype.into());
551	}
552	for (key, value) in [
553	("size", info.size_bytes),
554	("w", info.width),
555	("h", info.height),
556	] {
557	if let Some(value) = value {
558	out.insert(key.to_owned(), value.into());
559	}
560	}
561	out.into()
562	}
563
564	fn mxc(value: &JsonValue) -> Option<OwnedMxcUri> {
565	let uri = OwnedMxcUri::from(value.as_str()?);
566	uri.is_valid().then_some(uri)
567	}
568
569	fn object(value: JsonValue) -> JsonObject {
570	match value {
571	JsonValue::Object(object) => object,
572	_ => JsonObject::new(),
573	}
574	}
575
576	fn parse(json: &str) -> Result<JsonObject, Error> {
577	serde_json::from_str(json)
578	.map_err(\|source\| Error::with_source(ErrorKind::Server, "the pack is malformed", source))
579	}
580
581	fn raw<T>(content: &JsonObject) -> Raw<T> {
582	Raw::new(content)
583	.expect("a json object always serialises")
584	.cast_unchecked()
585	}