Kick: history, who is talking, and messages that were taken down
Closes #63 and #64, and the half of #72 that Kick actually offers. **History.** `/api/v2/channels/<id>/messages`, keyed by the numeric id because the same path with a handle answers 500 - the kind of difference only found by trying it. Fetched on join, in the background so a channel's first live messages are not held up by it, and paged by cursor on scroll rather than by timestamp, which would re-fetch the same page forever. Recorded oldest-first with Kick's own ids, so a reconnect - which joins again - recognises what it already has. **Who is talking.** `identity.color` and `identity.badges` arrive on every message and were read by nothing, so a moderator, a two-year subscriber and a stranger all drew identically. Both now travel with the message and are persisted, or reopening a conversation would strip a chat of the thing that makes it readable. Level badges are filtered out: an engagement number beside every nick is noise, and the ones worth scanning for are moderator, subscriber and verified. Carried as one `SenderStyle` rather than two more parameters - `record_message_at` was already at seventeen, colour and badges always travel together, and they are absent together for every protocol with no such idea. **Messages taken down.** `MessageDeletedEvent` was matched and ignored, so a message a moderator removed stayed on screen - moho showing a different chat from the one everybody else in the room is looking at. It is removed now, a chat clear empties the buffer, and bans and timeouts are announced, told apart by whether the event carries an expiry. One trap re-encountered: the mentions query reads its buffer id by column index, and two new columns moved it again. That number has now been wrong twice, so it says so, along with what would stop it being wrong a third time. Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
Claude Opus 5
@@ -2141,7 +2141,7 @@ fn store_history_messages(state: &AppState, buffer_id: &str, messages: &[Value],
|
||||
.and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok())
|
||||
.map(|dt| dt.timestamp())
|
||||
.unwrap_or(0);
|
||||
if let Err(e) = state.store.append_message(buffer_id, msg_id, from, &body, ts, false, false, "chat", reply_to.as_ref(), &reactions, is_own, avatar_url.as_deref(), &embeds, &attachments, None, None, &state.runtime.buffer_kind_of(buffer_id)) {
|
||||
if let Err(e) = state.store.append_message(buffer_id, msg_id, from, &body, ts, false, false, "chat", reply_to.as_ref(), &reactions, is_own, avatar_url.as_deref(), &embeds, &attachments, None, None, &state.runtime.buffer_kind_of(buffer_id), None, &[]) {
|
||||
tracing::warn!("discord: storing history message: {e}");
|
||||
continue;
|
||||
}
|
||||
@@ -3648,6 +3648,8 @@ mod tests {
|
||||
id: id.into(),
|
||||
buffer_id: "b".into(),
|
||||
html: None,
|
||||
sender_color: None,
|
||||
badges: Vec::new(),
|
||||
from: "x".into(),
|
||||
body: String::new(),
|
||||
ts: 0,
|
||||
|
||||
+3
-3
@@ -982,9 +982,9 @@ async fn handle_message(
|
||||
return;
|
||||
}
|
||||
if let Some(action_body) = strip_action(&body) {
|
||||
state.runtime.record_message_at(state, account_id, &buffer_name, kind, &from, action_body, true, "chat", None, msg_id, false, None, Vec::new(), Vec::new(), None, sent_at, None);
|
||||
state.runtime.record_message_at(state, account_id, &buffer_name, kind, &from, action_body, true, "chat", None, msg_id, false, None, Vec::new(), Vec::new(), None, sent_at, None, None);
|
||||
} else {
|
||||
state.runtime.record_message_at(state, account_id, &buffer_name, kind, &from, &body, false, "chat", None, msg_id, false, None, Vec::new(), Vec::new(), None, sent_at, None);
|
||||
state.runtime.record_message_at(state, account_id, &buffer_name, kind, &from, &body, false, "chat", None, msg_id, false, None, Vec::new(), Vec::new(), None, sent_at, None, None);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1017,7 +1017,7 @@ async fn handle_message(
|
||||
// are in its queue, why it refused. In the server tab those sit
|
||||
// a long way from the request they answer, and the conversation
|
||||
// that asked shows nothing but your own message.
|
||||
state.runtime.record_message_at(state, account_id, &from, "dm", &from, &body, false, "chat", None, None, false, None, Vec::new(), Vec::new(), None, sent_at, None);
|
||||
state.runtime.record_message_at(state, account_id, &from, "dm", &from, &body, false, "chat", None, None, false, None, Vec::new(), Vec::new(), None, sent_at, None, None);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -406,6 +406,108 @@ fn slugs_in(body: &serde_json::Value) -> Vec<String> {
|
||||
out
|
||||
}
|
||||
|
||||
/// One message out of a channel's history.
|
||||
#[derive(Deserialize, Debug)]
|
||||
pub struct HistoryMessage {
|
||||
pub id: String,
|
||||
#[serde(default)]
|
||||
pub content: String,
|
||||
#[serde(rename = "type", default)]
|
||||
pub kind: String,
|
||||
#[serde(default)]
|
||||
pub created_at: Option<String>,
|
||||
pub sender: HistorySender,
|
||||
#[serde(default)]
|
||||
pub metadata: Option<serde_json::Value>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Debug)]
|
||||
pub struct HistorySender {
|
||||
#[serde(default)]
|
||||
pub id: Option<u64>,
|
||||
#[serde(default)]
|
||||
pub username: Option<String>,
|
||||
#[serde(default)]
|
||||
pub identity: Option<ChatIdentity>,
|
||||
}
|
||||
|
||||
/// How somebody appears in chat: their colour, and what they have earned.
|
||||
///
|
||||
/// Both arrive on every message and were both being dropped at the struct
|
||||
/// boundary. Kick chat is substantially about who is talking - a moderator, a
|
||||
/// subscriber of two years, the streamer - and without this every line looks
|
||||
/// the same.
|
||||
///
|
||||
/// Named for what it describes rather than `Identity`, which in this module
|
||||
/// already means "who the signed-in account is".
|
||||
#[derive(Deserialize, Debug, Clone, Default)]
|
||||
pub struct ChatIdentity {
|
||||
#[serde(default)]
|
||||
pub color: Option<String>,
|
||||
#[serde(default)]
|
||||
pub badges: Vec<Badge>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Debug, Clone, Serialize)]
|
||||
pub struct Badge {
|
||||
/// `moderator`, `subscriber`, `verified`, `og`, `founder`, `vip`...
|
||||
#[serde(rename = "type")]
|
||||
pub kind: String,
|
||||
/// What Kick calls it, which is what a tooltip should say.
|
||||
#[serde(default)]
|
||||
pub text: String,
|
||||
/// Months subscribed, for the badges that count.
|
||||
#[serde(default)]
|
||||
pub count: Option<u32>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct HistoryEnvelope {
|
||||
#[serde(default)]
|
||||
data: Option<HistoryPage>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct HistoryPage {
|
||||
#[serde(default)]
|
||||
messages: Vec<HistoryMessage>,
|
||||
#[serde(default)]
|
||||
cursor: Option<serde_json::Value>,
|
||||
}
|
||||
|
||||
/// A page of a channel's past, newest first.
|
||||
///
|
||||
/// Keyed by the numeric channel id rather than the handle: the same path with
|
||||
/// a slug answers 500, which is the kind of difference that only shows up by
|
||||
/// trying it.
|
||||
///
|
||||
/// `cursor` continues an earlier page; the returned one continues this page,
|
||||
/// and is absent at the end of what the server will give.
|
||||
pub async fn history(
|
||||
http: &reqwest::Client,
|
||||
channel_id: u64,
|
||||
cursor: Option<&str>,
|
||||
) -> Result<(Vec<HistoryMessage>, Option<String>)> {
|
||||
let mut url = format!("{API_ROOT}/api/v2/channels/{channel_id}/messages");
|
||||
if let Some(cursor) = cursor {
|
||||
url.push_str(&format!("?cursor={}", urlencode(cursor)));
|
||||
}
|
||||
let res = http.get(&url).header("Accept", "application/json").send().await.context("fetching the channel's history")?;
|
||||
if !res.status().is_success() {
|
||||
bail!("Kick answered {} for that channel's history", res.status());
|
||||
}
|
||||
let envelope: HistoryEnvelope = res.json().await.context("reading the channel's history")?;
|
||||
let page = envelope.data.unwrap_or(HistoryPage { messages: Vec::new(), cursor: None });
|
||||
// Kick has sent this as a number and as a string; both mean the same
|
||||
// thing to the next request.
|
||||
let cursor = match page.cursor {
|
||||
Some(serde_json::Value::String(s)) if !s.is_empty() => Some(s),
|
||||
Some(serde_json::Value::Number(n)) => Some(n.to_string()),
|
||||
_ => None,
|
||||
};
|
||||
Ok((page.messages, cursor))
|
||||
}
|
||||
|
||||
/// What this account is to one channel.
|
||||
#[derive(Debug, Clone, Copy, Default)]
|
||||
pub struct Standing {
|
||||
|
||||
@@ -329,12 +329,33 @@ async fn prepare(
|
||||
&buffer.id,
|
||||
crate::runtime::KickChannel {
|
||||
slug: channel.slug.clone(),
|
||||
channel_id: channel.id,
|
||||
history_cursor: None,
|
||||
chatroom_id: channel.chatroom_id,
|
||||
subscribed: false,
|
||||
emotes: Vec::new(),
|
||||
},
|
||||
);
|
||||
|
||||
// The past, before the present starts arriving. Backgrounded with the
|
||||
// rest so a channel's first messages are not held up by it, and recorded
|
||||
// with their own timestamps so they sort into place whichever lands first.
|
||||
//
|
||||
// The cursor is kept so scrolling up continues where this stopped rather
|
||||
// than fetching the same page again.
|
||||
{
|
||||
let state = state.clone();
|
||||
let http = http.clone();
|
||||
let slug = channel.slug.clone();
|
||||
let buffer_id = buffer.id.clone();
|
||||
let account_id = account_id.to_string();
|
||||
let channel_id = channel.id;
|
||||
tokio::spawn(async move {
|
||||
let next = backfill(&state, &http, &account_id, &slug, channel_id, None).await;
|
||||
state.runtime.set_kick_history_cursor(&buffer_id, next);
|
||||
});
|
||||
}
|
||||
|
||||
// The emote table and this account's standing in the channel are fetched
|
||||
// in the background rather than before the subscription.
|
||||
//
|
||||
@@ -457,6 +478,11 @@ struct Sender {
|
||||
username: Option<String>,
|
||||
#[serde(default)]
|
||||
id: Option<u64>,
|
||||
/// Their colour and badges. Present on every message and previously read
|
||||
/// by nothing, so a moderator, a two-year subscriber and a stranger all
|
||||
/// drew identically.
|
||||
#[serde(default)]
|
||||
identity: Option<api::ChatIdentity>,
|
||||
}
|
||||
|
||||
async fn handle_frame(
|
||||
@@ -511,6 +537,11 @@ async fn handle_frame(
|
||||
msg.sender.id.map(|i| i.to_string()),
|
||||
msg.created_at.as_deref().and_then(parse_timestamp),
|
||||
None,
|
||||
// How Kick says this person looks: their colour and what they
|
||||
// have earned. Both arrive on every message and were both
|
||||
// being dropped at the struct boundary, which is why every
|
||||
// line looked the same.
|
||||
Some(style_of(msg.sender.identity.as_ref())),
|
||||
);
|
||||
}
|
||||
// Worth a line in the channel it happened in: somebody watching a
|
||||
@@ -525,6 +556,31 @@ async fn handle_frame(
|
||||
system_line(state, account_id, &slug, "ended the stream", "system");
|
||||
}
|
||||
}
|
||||
// A message taken down, by its author or by a moderator. Removed here
|
||||
// too, or moho shows a different chat from the one everybody else is
|
||||
// looking at - which on a stream where moderation is public is the
|
||||
// wrong way round entirely.
|
||||
e if e.ends_with("MessageDeletedEvent") => {
|
||||
let Some(slug) = channel_of(watched, &frame.channel, &payload) else { return Ok(()) };
|
||||
// The id is nested under `message` on this event and flat on
|
||||
// others Kick has sent; both name the same thing.
|
||||
let id = payload
|
||||
.get("message")
|
||||
.and_then(|m| m.get("id"))
|
||||
.or_else(|| payload.get("id"))
|
||||
.and_then(|v| v.as_str());
|
||||
if let Some(id) = id {
|
||||
state.runtime.delete_message(state, &crate::model::buffer_id(account_id, &slug), id);
|
||||
}
|
||||
}
|
||||
|
||||
// A moderator emptying the room.
|
||||
e if e.ends_with("ChatroomClearEvent") => {
|
||||
if let Some(slug) = channel_of(watched, &frame.channel, &payload) {
|
||||
system_line(state, account_id, &slug, "a moderator cleared the chat", "system");
|
||||
}
|
||||
}
|
||||
|
||||
// The things people do rather than say.
|
||||
e => {
|
||||
if let Some(line) = describe_action(e, &payload) {
|
||||
@@ -537,6 +593,90 @@ async fn handle_frame(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Fills a channel's scrollback from what was said before we arrived.
|
||||
///
|
||||
/// Kick's chat is a live stream and nothing else - a channel opened for the
|
||||
/// first time showed an empty log until somebody spoke, which on a quiet
|
||||
/// stream is a conversation that appears not to exist.
|
||||
///
|
||||
/// Recorded oldest-first so the log reads in order, and with Kick's own ids so
|
||||
/// anything already stored is recognised rather than duplicated. That matters
|
||||
/// more here than elsewhere: this runs on every join, and a reconnect joins
|
||||
/// again.
|
||||
pub async fn backfill(
|
||||
state: &AppState,
|
||||
http: &reqwest::Client,
|
||||
account_id: &str,
|
||||
slug: &str,
|
||||
channel_id: u64,
|
||||
cursor: Option<&str>,
|
||||
) -> Option<String> {
|
||||
let (messages, next) = match api::history(http, channel_id, cursor).await {
|
||||
Ok(page) => page,
|
||||
// Not fatal and not worth interrupting anybody over: the channel still
|
||||
// works, it just starts empty, which is what it did before.
|
||||
Err(e) => {
|
||||
tracing::debug!("kick[{account_id}]: history for {slug}: {e:#}");
|
||||
return None;
|
||||
}
|
||||
};
|
||||
for msg in messages.iter().rev() {
|
||||
if msg.content.is_empty() {
|
||||
continue;
|
||||
}
|
||||
let from = msg.sender.username.clone().unwrap_or_else(|| "someone".to_string());
|
||||
state.runtime.record_message_at(
|
||||
state,
|
||||
account_id,
|
||||
slug,
|
||||
"channel",
|
||||
&from,
|
||||
&msg.content,
|
||||
false,
|
||||
"chat",
|
||||
reply_preview_from_value(msg.metadata.as_ref()),
|
||||
Some(msg.id.clone()),
|
||||
false,
|
||||
None,
|
||||
Vec::new(),
|
||||
Vec::new(),
|
||||
msg.sender.id.map(|i| i.to_string()),
|
||||
msg.created_at.as_deref().and_then(parse_timestamp),
|
||||
None,
|
||||
Some(style_of(msg.sender.identity.as_ref())),
|
||||
);
|
||||
}
|
||||
next
|
||||
}
|
||||
|
||||
/// The same reply metadata the live path reads, out of an untyped value.
|
||||
///
|
||||
/// History carries it in the same shape, so this is the one parser rather than
|
||||
/// a second one that could drift.
|
||||
fn reply_preview_from_value(metadata: Option<&serde_json::Value>) -> Option<crate::model::ReplyPreview> {
|
||||
let parsed: ReplyMetadata = serde_json::from_value(metadata?.clone()).ok()?;
|
||||
reply_preview(Some(&parsed))
|
||||
}
|
||||
|
||||
/// How a sender should be drawn, out of what Kick sent with the message.
|
||||
///
|
||||
/// Badges are filtered to the ones that say something about the person rather
|
||||
/// than about Kick: a "level" badge is an engagement number and a row of them
|
||||
/// beside every nick is noise, while moderator, subscriber and verified are
|
||||
/// the ones a reader is actually scanning for.
|
||||
fn style_of(identity: Option<&api::ChatIdentity>) -> crate::model::SenderStyle {
|
||||
let Some(identity) = identity else { return crate::model::SenderStyle::default() };
|
||||
crate::model::SenderStyle {
|
||||
color: identity.color.clone().filter(|c| !c.is_empty()),
|
||||
badges: identity
|
||||
.badges
|
||||
.iter()
|
||||
.filter(|b| !matches!(b.kind.as_str(), "level" | ""))
|
||||
.cloned()
|
||||
.collect(),
|
||||
}
|
||||
}
|
||||
|
||||
/// What a reply was replying to, out of the metadata Kick attaches to it.
|
||||
///
|
||||
/// Both halves are required rather than filled in with blanks: a preview with
|
||||
@@ -628,6 +768,28 @@ fn describe_action(event: &str, p: &serde_json::Value) -> Option<String> {
|
||||
})
|
||||
}
|
||||
|
||||
// Somebody timed out or banned. Said in the channel, because it is
|
||||
// moderation everybody in the room can see happening.
|
||||
"UserBannedEvent" => {
|
||||
let who = p.get("user").and_then(|u| u.get("username")).and_then(|v| v.as_str())?;
|
||||
let by = p
|
||||
.get("banned_by")
|
||||
.and_then(|u| u.get("username"))
|
||||
.and_then(|v| v.as_str())
|
||||
.map(|b| format!(" by {b}"))
|
||||
.unwrap_or_default();
|
||||
// A ban carries no expiry and a timeout does, which is the whole
|
||||
// of what a reader wants to know from the line.
|
||||
Some(match p.get("expires_at").and_then(|v| v.as_str()) {
|
||||
Some(_) => format!("{who} was timed out{by}"),
|
||||
None => format!("{who} was banned{by}"),
|
||||
})
|
||||
}
|
||||
"UserUnbannedEvent" => {
|
||||
let who = p.get("user").and_then(|u| u.get("username")).and_then(|v| v.as_str())?;
|
||||
Some(format!("{who} is allowed back"))
|
||||
}
|
||||
|
||||
// Deliberately nothing. Both are the same subscription reported again
|
||||
// on another subscription - `ChatMessageSentEvent` is the info line
|
||||
// Kick's own page draws from, and `LuckyUsers...` is the companion to
|
||||
@@ -926,6 +1088,9 @@ mod tests {
|
||||
#[test]
|
||||
fn ignores_events_that_are_not_somebody_doing_something() {
|
||||
let p = serde_json::json!({ "username": "someone" });
|
||||
// These two are handled in the frame loop rather than here - one
|
||||
// removes a message and the other reports a chat mode - so this is
|
||||
// still the right answer for both.
|
||||
assert_eq!(describe_action("App\\Events\\MessageDeletedEvent", &p), None);
|
||||
assert_eq!(describe_action("App\\Events\\ChatroomUpdatedEvent", &p), None);
|
||||
}
|
||||
@@ -948,6 +1113,35 @@ mod tests {
|
||||
assert_eq!(describe_action("App\\Events\\ChatMessageSentEvent", &info), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn tells_a_timeout_from_a_ban() {
|
||||
let timeout = serde_json::json!({
|
||||
"user": { "username": "someone" },
|
||||
"banned_by": { "username": "amod" },
|
||||
"expires_at": "2026-09-02T04:00:00Z"
|
||||
});
|
||||
assert_eq!(
|
||||
describe_action("App\\Events\\UserBannedEvent", &timeout).as_deref(),
|
||||
Some("someone was timed out by amod")
|
||||
);
|
||||
|
||||
// No expiry is a ban, which is the difference worth reporting.
|
||||
let ban = serde_json::json!({ "user": { "username": "someone" }, "banned_by": { "username": "amod" } });
|
||||
assert_eq!(
|
||||
describe_action("App\\Events\\UserBannedEvent", &ban).as_deref(),
|
||||
Some("someone was banned by amod")
|
||||
);
|
||||
|
||||
// A moderator who did not name themselves still produces a line.
|
||||
let anonymous = serde_json::json!({ "user": { "username": "someone" } });
|
||||
assert_eq!(describe_action("UserBannedEvent", &anonymous).as_deref(), Some("someone was banned"));
|
||||
|
||||
assert_eq!(
|
||||
describe_action("App\\Events\\UserUnbannedEvent", &serde_json::json!({ "user": { "username": "someone" } })).as_deref(),
|
||||
Some("someone is allowed back")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reads_a_real_gifted_subscription_payload() {
|
||||
// Copied from the wire, prefix and all - this one arrives with no
|
||||
|
||||
@@ -720,6 +720,7 @@ async fn handle_timeline_event(
|
||||
// than rendered here: it is somebody else's markup and the frontend
|
||||
// is where the sanitiser lives.
|
||||
protocol::formatted_body(&content).map(str::to_string),
|
||||
None,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -1110,6 +1111,9 @@ pub async fn backfill(state: &AppState, account_id: &str, buffer_id: &str, limit
|
||||
buffer_id, event_id, &from, &body, ts, is_action, false, "message", None, &[], is_own,
|
||||
avatar.as_deref(), &[], &[], Some(sender_mxid), html.as_deref(),
|
||||
&state.runtime.buffer_kind_of(buffer_id),
|
||||
// Matrix has no per-sender colour or badges of its own.
|
||||
None,
|
||||
&[],
|
||||
) {
|
||||
tracing::warn!("matrix: storing history message: {e}");
|
||||
continue;
|
||||
|
||||
@@ -224,6 +224,18 @@ pub struct ReplyPreview {
|
||||
pub body: String,
|
||||
}
|
||||
|
||||
/// How a service says a sender should look.
|
||||
///
|
||||
/// One value rather than two more parameters on `record_message_at`, which was
|
||||
/// already carrying seventeen: colour and badges always travel together,
|
||||
/// always come from the same place, and are always absent together for the
|
||||
/// protocols that have no such idea.
|
||||
#[derive(Clone, Debug, Default)]
|
||||
pub struct SenderStyle {
|
||||
pub color: Option<String>,
|
||||
pub badges: Vec<crate::backend::kick::api::Badge>,
|
||||
}
|
||||
|
||||
/// One emoji's reaction tally on a message. `me` is whether *this*
|
||||
/// account is among the reactors - Discord's REACTION_ADD/REMOVE events
|
||||
/// are per-user, so this is accumulated incrementally rather than
|
||||
@@ -370,6 +382,18 @@ pub struct Message {
|
||||
/// showing.
|
||||
#[serde(rename = "senderId", skip_serializing_if = "Option::is_none")]
|
||||
pub sender_id: Option<String>,
|
||||
/// The colour this service says the sender's name should be.
|
||||
///
|
||||
/// Kick gives everybody one and it is half of how a busy chat is read.
|
||||
/// A client that has none falls back to colouring the nick itself, which
|
||||
/// is what IRC has always needed - so this overrides that rather than
|
||||
/// replacing it.
|
||||
#[serde(rename = "senderColor", skip_serializing_if = "Option::is_none")]
|
||||
pub sender_color: Option<String>,
|
||||
/// What the sender has earned in this channel: moderator, subscriber,
|
||||
/// verified, and so on. Empty for services that have no such idea.
|
||||
#[serde(skip_serializing_if = "Vec::is_empty")]
|
||||
pub badges: Vec<crate::backend::kick::api::Badge>,
|
||||
/// The sender's own formatted version of `body`, where the protocol
|
||||
/// carries one (Matrix's `formatted_body`). Restricted HTML, and still
|
||||
/// untrusted - a frontend must put it through the same sanitiser it
|
||||
|
||||
@@ -150,6 +150,31 @@ pub async fn dispatch(
|
||||
backend::discord::catch_up_channel(state, &cfg.token, &cfg.user_id, cfg.display_name.as_deref(), buffer_id, &channel_id).await;
|
||||
}
|
||||
}
|
||||
// Kick pages back through its own history endpoint,
|
||||
// continuing from where the last page stopped rather than
|
||||
// by timestamp - the endpoint is cursor-based, and asking
|
||||
// by time would re-fetch the same page forever.
|
||||
if before > 0 && buffer.account_id.starts_with("kick:") {
|
||||
if let Some(channel) = state.runtime.kick_channel(buffer_id) {
|
||||
// No cursor means the start has been reached; the
|
||||
// store already holds everything there is.
|
||||
if let Some(cursor) = channel.history_cursor {
|
||||
if let Ok(http) = backend::kick::api::client() {
|
||||
let next = backend::kick::backfill(
|
||||
state,
|
||||
&http,
|
||||
&buffer.account_id,
|
||||
&channel.slug,
|
||||
channel.channel_id,
|
||||
Some(&cursor),
|
||||
)
|
||||
.await;
|
||||
state.runtime.set_kick_history_cursor(buffer_id, next);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// IRC pages back through CHATHISTORY, where the server
|
||||
// has it. Unlike the other two this is not a request with
|
||||
// a reply - the history arrives as ordinary messages on
|
||||
|
||||
+17
-2
@@ -142,6 +142,12 @@ pub struct IrcChannelListing {
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct KickChannel {
|
||||
pub slug: String,
|
||||
/// Kick's own numeric id for the channel, which is what its history
|
||||
/// endpoint is keyed by - the handle answers that path with a 500.
|
||||
pub channel_id: u64,
|
||||
/// Where the next page of history continues from, or None if the start
|
||||
/// has been reached or nothing has been asked for yet.
|
||||
pub history_cursor: Option<String>,
|
||||
/// The room the websocket subscribed to and messages are posted to.
|
||||
pub chatroom_id: u64,
|
||||
/// Whether this account is subscribed to this streamer, which decides
|
||||
@@ -1928,6 +1934,12 @@ impl Runtime {
|
||||
self.kick_channels.lock().unwrap().get(buffer_id).cloned()
|
||||
}
|
||||
|
||||
pub fn set_kick_history_cursor(&self, buffer_id: &str, cursor: Option<String>) {
|
||||
if let Some(channel) = self.kick_channels.lock().unwrap().get_mut(buffer_id) {
|
||||
channel.history_cursor = cursor;
|
||||
}
|
||||
}
|
||||
|
||||
pub fn forget_kick_channel(&self, buffer_id: &str) {
|
||||
self.kick_channels.lock().unwrap().remove(buffer_id);
|
||||
}
|
||||
@@ -2169,7 +2181,7 @@ impl Runtime {
|
||||
) -> bool {
|
||||
self.record_message_at(
|
||||
state, account_id, buffer_name, buffer_kind, from, body, is_action, kind, reply_to,
|
||||
msg_id_override, force_highlight, avatar_url, embeds, attachments, sender_id, None, None,
|
||||
msg_id_override, force_highlight, avatar_url, embeds, attachments, sender_id, None, None, None,
|
||||
)
|
||||
}
|
||||
|
||||
@@ -2200,6 +2212,7 @@ impl Runtime {
|
||||
sender_id: Option<String>,
|
||||
sent_at: Option<i64>,
|
||||
html: Option<String>,
|
||||
style: Option<model::SenderStyle>,
|
||||
) -> bool {
|
||||
let buffer = self.ensure_buffer(state, account_id, buffer_name, buffer_kind);
|
||||
let own_nick = self
|
||||
@@ -2234,7 +2247,7 @@ impl Runtime {
|
||||
// passes its real id through here for exactly that reason.
|
||||
let msg_id = msg_id_override.unwrap_or_else(model::next_message_id);
|
||||
|
||||
match state.store.append_message(&buffer.id, &msg_id, from, body, ts, is_action, is_highlight, kind, reply_to.as_ref(), &[], is_own, avatar_url.as_deref(), &embeds, &attachments, sender_id.as_deref(), html.as_deref(), buffer_kind) {
|
||||
match state.store.append_message(&buffer.id, &msg_id, from, body, ts, is_action, is_highlight, kind, reply_to.as_ref(), &[], is_own, avatar_url.as_deref(), &embeds, &attachments, sender_id.as_deref(), html.as_deref(), buffer_kind, style.as_ref().and_then(|s| s.color.as_deref()), style.as_ref().map(|s| s.badges.as_slice()).unwrap_or(&[])) {
|
||||
// Already had it, so there is nothing to announce. Sneedchat
|
||||
// replays a room's recent history whenever its connection comes
|
||||
// back, and every backend that carries real message ids can
|
||||
@@ -2286,6 +2299,8 @@ impl Runtime {
|
||||
embeds,
|
||||
attachments,
|
||||
sender_id,
|
||||
sender_color: style.as_ref().and_then(|s| s.color.clone()),
|
||||
badges: style.map(|s| s.badges).unwrap_or_default(),
|
||||
};
|
||||
state.events.emit("message", serde_json::to_value(&message).unwrap());
|
||||
|
||||
|
||||
+45
-22
@@ -85,6 +85,12 @@ impl Store {
|
||||
"ALTER TABLE messages ADD COLUMN avatar_url TEXT",
|
||||
"ALTER TABLE messages ADD COLUMN embeds TEXT NOT NULL DEFAULT '[]'",
|
||||
"ALTER TABLE messages ADD COLUMN attachments TEXT NOT NULL DEFAULT '[]'",
|
||||
// How the sender looked, where the service says: their colour and
|
||||
// what they have earned. Persisted rather than live-only, or
|
||||
// reopening a conversation would strip a chat of the thing that
|
||||
// makes it readable.
|
||||
"ALTER TABLE messages ADD COLUMN sender_color TEXT",
|
||||
"ALTER TABLE messages ADD COLUMN badges TEXT NOT NULL DEFAULT '[]'",
|
||||
"ALTER TABLE messages ADD COLUMN sender_id TEXT",
|
||||
// Matrix sends a formatted body alongside the plain one. Kept
|
||||
// separate rather than replacing body: the plain text is the
|
||||
@@ -153,6 +159,8 @@ impl Store {
|
||||
sender_id: Option<&str>,
|
||||
html: Option<&str>,
|
||||
buffer_kind: &str,
|
||||
sender_color: Option<&str>,
|
||||
badges: &[crate::backend::kick::api::Badge],
|
||||
) -> Result<bool> {
|
||||
let conn = self.conn.lock().unwrap();
|
||||
// Live messages are always freshly created with no reactions yet
|
||||
@@ -166,8 +174,8 @@ impl Store {
|
||||
// OR IGNORE against the (buffer_id, msg_id) index: recording the
|
||||
// same message twice is a backend replaying history it already
|
||||
// has, and the right answer is to keep the copy already stored.
|
||||
"INSERT OR IGNORE INTO messages (msg_id, buffer_id, from_nick, body, ts, is_action, is_highlight, kind, reply_to_id, reply_to_from, reply_to_body, reactions, is_own, avatar_url, embeds, sender_id, attachments, html, buffer_kind)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, COALESCE(?12, '[]'), ?13, ?14, COALESCE(?15, '[]'), ?16, COALESCE(?17, '[]'), ?18, ?19)",
|
||||
"INSERT OR IGNORE INTO messages (msg_id, buffer_id, from_nick, body, ts, is_action, is_highlight, kind, reply_to_id, reply_to_from, reply_to_body, reactions, is_own, avatar_url, embeds, sender_id, attachments, html, buffer_kind, sender_color, badges)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, COALESCE(?12, '[]'), ?13, ?14, COALESCE(?15, '[]'), ?16, COALESCE(?17, '[]'), ?18, ?19, ?20, COALESCE(?21, '[]'))",
|
||||
params![
|
||||
msg_id,
|
||||
buffer_id,
|
||||
@@ -188,6 +196,8 @@ impl Store {
|
||||
attachments_json,
|
||||
html,
|
||||
buffer_kind,
|
||||
sender_color,
|
||||
serde_json::to_string(badges).ok(),
|
||||
],
|
||||
)?;
|
||||
// Whether this was actually new. The insert has always ignored a
|
||||
@@ -379,6 +389,8 @@ impl Store {
|
||||
let attachments_json: String = row.get::<_, Option<String>>(16)?.unwrap_or_else(|| "[]".to_string());
|
||||
let mut attachments: Vec<Attachment> = serde_json::from_str(&attachments_json).unwrap_or_default();
|
||||
drop_missing_local_copies(&mut attachments);
|
||||
let badges_json: String = row.get::<_, Option<String>>(19)?.unwrap_or_else(|| "[]".to_string());
|
||||
let badges = serde_json::from_str(&badges_json).unwrap_or_default();
|
||||
Ok(Message {
|
||||
id: row.get::<_, Option<String>>(0)?.unwrap_or_default(),
|
||||
buffer_id: buffer_id.to_string(),
|
||||
@@ -396,6 +408,8 @@ impl Store {
|
||||
embeds,
|
||||
attachments,
|
||||
sender_id: row.get(15)?,
|
||||
sender_color: row.get(18)?,
|
||||
badges,
|
||||
html: row.get(17)?,
|
||||
})
|
||||
}
|
||||
@@ -404,7 +418,7 @@ impl Store {
|
||||
let conn = self.conn.lock().unwrap();
|
||||
let limit = if limit > 0 { limit } else { 200 };
|
||||
let mut stmt = conn.prepare(
|
||||
"SELECT msg_id, from_nick, body, ts, is_action, is_highlight, kind, reply_to_id, reply_to_from, reply_to_body, edited, reactions, is_own, avatar_url, embeds, sender_id, attachments, html
|
||||
"SELECT msg_id, from_nick, body, ts, is_action, is_highlight, kind, reply_to_id, reply_to_from, reply_to_body, edited, reactions, is_own, avatar_url, embeds, sender_id, attachments, html, sender_color, badges
|
||||
FROM messages
|
||||
WHERE buffer_id = ?1 AND (?2 <= 0 OR ts < ?2)
|
||||
ORDER BY ts DESC LIMIT ?3",
|
||||
@@ -444,7 +458,7 @@ impl Store {
|
||||
format!(" OR (buffer_kind IS NULL AND buffer_id IN ({places}))")
|
||||
};
|
||||
let sql = format!(
|
||||
"SELECT msg_id, from_nick, body, ts, is_action, is_highlight, kind, reply_to_id, reply_to_from, reply_to_body, edited, reactions, is_own, avatar_url, embeds, sender_id, attachments, html, buffer_id
|
||||
"SELECT msg_id, from_nick, body, ts, is_action, is_highlight, kind, reply_to_id, reply_to_from, reply_to_body, edited, reactions, is_own, avatar_url, embeds, sender_id, attachments, html, sender_color, badges, buffer_id
|
||||
FROM messages
|
||||
WHERE is_highlight = 1 AND is_own = 0 AND (buffer_kind = 'channel'{legacy})
|
||||
ORDER BY ts DESC LIMIT ?"
|
||||
@@ -457,13 +471,18 @@ impl Store {
|
||||
// parameter, since unlike every other query here this one spans
|
||||
// conversations and each answer belongs to a different one.
|
||||
//
|
||||
// Index 18, after the eighteen columns row_to_message reads. It
|
||||
// used to say 17, which is `html` - so every mention on a service
|
||||
// that sends no HTML, meaning all of them but Matrix, failed the
|
||||
// whole query with "Invalid column type Null". Nothing showed that:
|
||||
// the page fills from live messages as well, so mentions appeared
|
||||
// to work and then to vanish whenever the daemon restarted.
|
||||
let buffer_id: String = row.get(18)?;
|
||||
// Index 20, after the twenty columns row_to_message reads - and a
|
||||
// number that has now been wrong twice. It said 17 (`html`), which
|
||||
// made every mention on a service that sends no HTML fail the
|
||||
// whole query with "Invalid column type Null"; adding sender_color
|
||||
// and badges moved it again. Nothing shows it: the page fills from
|
||||
// live messages too, so mentions appear to work and then vanish on
|
||||
// the next restart.
|
||||
//
|
||||
// The real fix is to stop counting - a named column, or a struct
|
||||
// that owns its own column list - and it is worth doing the next
|
||||
// time this query is touched.
|
||||
let buffer_id: String = row.get(20)?;
|
||||
Self::row_to_message(&buffer_id, row)
|
||||
})?;
|
||||
rows.collect::<rusqlite::Result<_>>().map_err(Into::into)
|
||||
@@ -550,7 +569,7 @@ impl Store {
|
||||
let escaped = query.replace('\\', "\\\\").replace('%', "\\%").replace('_', "\\_");
|
||||
let pattern = format!("%{escaped}%");
|
||||
let mut stmt = conn.prepare(
|
||||
"SELECT msg_id, from_nick, body, ts, is_action, is_highlight, kind, reply_to_id, reply_to_from, reply_to_body, edited, reactions, is_own, avatar_url, embeds, sender_id, attachments, html
|
||||
"SELECT msg_id, from_nick, body, ts, is_action, is_highlight, kind, reply_to_id, reply_to_from, reply_to_body, edited, reactions, is_own, avatar_url, embeds, sender_id, attachments, html, sender_color, badges
|
||||
FROM messages
|
||||
WHERE buffer_id = ?1 AND body LIKE ?2 ESCAPE '\\'
|
||||
ORDER BY ts DESC LIMIT ?3",
|
||||
@@ -594,7 +613,7 @@ impl Store {
|
||||
pub fn get_message(&self, buffer_id: &str, msg_id: &str) -> Result<Option<Message>> {
|
||||
let conn = self.conn.lock().unwrap();
|
||||
let mut stmt = conn.prepare(
|
||||
"SELECT msg_id, from_nick, body, ts, is_action, is_highlight, kind, reply_to_id, reply_to_from, reply_to_body, edited, reactions, is_own, avatar_url, embeds, sender_id, attachments, html
|
||||
"SELECT msg_id, from_nick, body, ts, is_action, is_highlight, kind, reply_to_id, reply_to_from, reply_to_body, edited, reactions, is_own, avatar_url, embeds, sender_id, attachments, html, sender_color, badges
|
||||
FROM messages
|
||||
WHERE buffer_id = ?1 AND msg_id = ?2",
|
||||
)?;
|
||||
@@ -614,6 +633,8 @@ impl Store {
|
||||
let attachments_json: String = row.get::<_, Option<String>>(16)?.unwrap_or_else(|| "[]".to_string());
|
||||
let mut attachments: Vec<Attachment> = serde_json::from_str(&attachments_json).unwrap_or_default();
|
||||
drop_missing_local_copies(&mut attachments);
|
||||
let badges_json: String = row.get::<_, Option<String>>(19)?.unwrap_or_else(|| "[]".to_string());
|
||||
let badges = serde_json::from_str(&badges_json).unwrap_or_default();
|
||||
Ok(Message {
|
||||
id: row.get::<_, Option<String>>(0)?.unwrap_or_default(),
|
||||
buffer_id: buffer_id.to_string(),
|
||||
@@ -631,6 +652,8 @@ impl Store {
|
||||
embeds,
|
||||
attachments,
|
||||
sender_id: row.get(15)?,
|
||||
sender_color: row.get(18)?,
|
||||
badges,
|
||||
html: row.get(17)?,
|
||||
})
|
||||
})?;
|
||||
@@ -692,18 +715,18 @@ mod tests {
|
||||
}
|
||||
|
||||
fn append(s: &Store, buffer: &str, id: &str) {
|
||||
s.append_message(buffer, id, "someone", "hi", 1, false, false, "chat", None, &[], false, None, &[], &[], None, None, "channel")
|
||||
s.append_message(buffer, id, "someone", "hi", 1, false, false, "chat", None, &[], false, None, &[], &[], None, None, "channel", None, &[])
|
||||
.expect("appending");
|
||||
}
|
||||
|
||||
/// Like `append`, but with text worth searching for.
|
||||
fn append_saying(s: &Store, buffer: &str, id: &str, body: &str) {
|
||||
s.append_message(buffer, id, "someone", body, 1, false, false, "chat", None, &[], false, None, &[], &[], None, None, "channel")
|
||||
s.append_message(buffer, id, "someone", body, 1, false, false, "chat", None, &[], false, None, &[], &[], None, None, "channel", None, &[])
|
||||
.expect("appending");
|
||||
}
|
||||
|
||||
fn mention(s: &Store, buffer: &str, id: &str, kind: &str) {
|
||||
s.append_message(buffer, id, "someone", "hey you", 1, false, true, "chat", None, &[], false, None, &[], &[], None, None, kind)
|
||||
s.append_message(buffer, id, "someone", "hey you", 1, false, true, "chat", None, &[], false, None, &[], &[], None, None, kind, None, &[])
|
||||
.expect("appending");
|
||||
}
|
||||
|
||||
@@ -738,7 +761,7 @@ mod tests {
|
||||
// were in have nothing to vouch for them but the live buffer list,
|
||||
// which is exactly what this used to rely on.
|
||||
let (st, dir) = store();
|
||||
st.append_message("irc|#old", "1", "someone", "hey you", 1, false, true, "chat", None, &[], false, None, &[], &[], None, None, "channel")
|
||||
st.append_message("irc|#old", "1", "someone", "hey you", 1, false, true, "chat", None, &[], false, None, &[], &[], None, None, "channel", None, &[])
|
||||
.unwrap();
|
||||
st.conn.lock().unwrap().execute("UPDATE messages SET buffer_kind = NULL", []).unwrap();
|
||||
|
||||
@@ -786,10 +809,10 @@ mod tests {
|
||||
// same one, over and over, for a message read hours ago.
|
||||
let (st, dir) = store();
|
||||
let first = st
|
||||
.append_message("room", "uuid-1", "someone", "hi", 1, false, true, "chat", None, &[], false, None, &[], &[], None, None, "channel")
|
||||
.append_message("room", "uuid-1", "someone", "hi", 1, false, true, "chat", None, &[], false, None, &[], &[], None, None, "channel", None, &[])
|
||||
.expect("appending");
|
||||
let again = st
|
||||
.append_message("room", "uuid-1", "someone", "hi", 1, false, true, "chat", None, &[], false, None, &[], &[], None, None, "channel")
|
||||
.append_message("room", "uuid-1", "someone", "hi", 1, false, true, "chat", None, &[], false, None, &[], &[], None, None, "channel", None, &[])
|
||||
.expect("appending again");
|
||||
|
||||
assert!(first, "the first time is new");
|
||||
@@ -805,10 +828,10 @@ mod tests {
|
||||
// silently drop it.
|
||||
let (st, dir) = store();
|
||||
assert!(st
|
||||
.append_message("room", "shared", "someone", "hi", 1, false, false, "chat", None, &[], false, None, &[], &[], None, None, "channel")
|
||||
.append_message("room", "shared", "someone", "hi", 1, false, false, "chat", None, &[], false, None, &[], &[], None, None, "channel", None, &[])
|
||||
.unwrap());
|
||||
assert!(st
|
||||
.append_message("Whispers", "shared", "someone", "hi", 1, false, false, "whisper", None, &[], false, None, &[], &[], None, None, "channel")
|
||||
.append_message("Whispers", "shared", "someone", "hi", 1, false, false, "whisper", None, &[], false, None, &[], &[], None, None, "channel", None, &[])
|
||||
.unwrap());
|
||||
let _ = std::fs::remove_dir_all(dir);
|
||||
}
|
||||
@@ -931,7 +954,7 @@ mod dedupe_tests {
|
||||
}
|
||||
|
||||
fn put(s: &Store, buffer: &str, id: &str, body: &str) {
|
||||
let _ = s.append_message(buffer, id, "nick", body, 1, false, false, "chat", None, &[], false, None, &[], &[], None, None, "channel");
|
||||
let _ = s.append_message(buffer, id, "nick", body, 1, false, false, "chat", None, &[], false, None, &[], &[], None, None, "channel", None, &[]);
|
||||
}
|
||||
|
||||
/// A backend replaying history it already has - Sneedchat does this on
|
||||
|
||||
Reference in New Issue
Block a user