From a15e3d580124f561864c6a39f1e035eb1b3aab13 Mon Sep 17 00:00:00 2001
From: Owen Jacobson <owen@grimoire.ca>
Date: Mon, 30 Jun 2025 22:00:57 -0400
Subject: Rename "channel" to "conversation" within the server.

I've split this from the schema and API changes because, frankly, it's huge. Annoyingly so. There are no semantic changes in this, it's all symbol changes, but there are a _lot_ of them because the term "channel" leaks all over everything in a service whose primary role is managing messages sent to channels (now, conversations).

I found a buggy test while working on this! It's not fixed in this commit, because it felt mean to hide a real change in the middle of this much chaff.
---
 src/event/handlers/stream/test/channel.rs      | 273 -------------------------
 src/event/handlers/stream/test/conversation.rs | 273 +++++++++++++++++++++++++
 src/event/handlers/stream/test/message.rs      |  65 +++---
 src/event/handlers/stream/test/mod.rs          |   2 +-
 src/event/handlers/stream/test/resume.rs       |  43 ++--
 src/event/handlers/stream/test/token.rs        |  24 +--
 6 files changed, 342 insertions(+), 338 deletions(-)
 delete mode 100644 src/event/handlers/stream/test/channel.rs
 create mode 100644 src/event/handlers/stream/test/conversation.rs

(limited to 'src/event/handlers/stream')

diff --git a/src/event/handlers/stream/test/channel.rs b/src/event/handlers/stream/test/channel.rs
deleted file mode 100644
index 2b87ce2..0000000
--- a/src/event/handlers/stream/test/channel.rs
+++ /dev/null
@@ -1,273 +0,0 @@
-use axum::extract::State;
-use axum_extra::extract::Query;
-use futures::{future, stream::StreamExt as _};
-
-use crate::test::fixtures::{self, future::Expect as _};
-
-#[tokio::test]
-async fn creating() {
-    // Set up the environment
-
-    let app = fixtures::scratch_app().await;
-    let resume_point = fixtures::boot::resume_point(&app).await;
-
-    // Subscribe
-
-    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
-    let super::Response(events) = super::handler(
-        State(app.clone()),
-        subscriber,
-        None,
-        Query(super::QueryParams { resume_point }),
-    )
-    .await
-    .expect("subscribe never fails");
-
-    // Create a channel
-
-    let name = fixtures::channel::propose();
-    let channel = app
-        .channels()
-        .create(&name, &fixtures::now())
-        .await
-        .expect("creating a channel succeeds");
-
-    // Verify channel created event
-
-    events
-        .filter_map(fixtures::event::stream::channel)
-        .filter_map(fixtures::event::stream::channel::created)
-        .filter(|event| future::ready(event.channel == channel))
-        .next()
-        .expect_some("channel created event is delivered")
-        .await;
-}
-
-#[tokio::test]
-async fn previously_created() {
-    // Set up the environment
-
-    let app = fixtures::scratch_app().await;
-    let resume_point = fixtures::boot::resume_point(&app).await;
-
-    // Create a channel
-
-    let name = fixtures::channel::propose();
-    let channel = app
-        .channels()
-        .create(&name, &fixtures::now())
-        .await
-        .expect("creating a channel succeeds");
-
-    // Subscribe
-
-    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
-    let super::Response(events) = super::handler(
-        State(app.clone()),
-        subscriber,
-        None,
-        Query(super::QueryParams { resume_point }),
-    )
-    .await
-    .expect("subscribe never fails");
-
-    // Verify channel created event
-
-    let _ = events
-        .filter_map(fixtures::event::stream::channel)
-        .filter_map(fixtures::event::stream::channel::created)
-        .filter(|event| future::ready(event.channel == channel))
-        .next()
-        .expect_some("channel created event is delivered")
-        .await;
-}
-
-#[tokio::test]
-async fn expiring() {
-    // Set up the environment
-
-    let app = fixtures::scratch_app().await;
-    let channel = fixtures::channel::create(&app, &fixtures::ancient()).await;
-    let resume_point = fixtures::boot::resume_point(&app).await;
-
-    // Subscribe
-
-    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
-    let super::Response(events) = super::handler(
-        State(app.clone()),
-        subscriber,
-        None,
-        Query(super::QueryParams { resume_point }),
-    )
-    .await
-    .expect("subscribe never fails");
-
-    // Expire channels
-
-    app.channels()
-        .expire(&fixtures::now())
-        .await
-        .expect("expiring channels always succeeds");
-
-    // Check for expiry event
-    let _ = events
-        .filter_map(fixtures::event::stream::channel)
-        .filter_map(fixtures::event::stream::channel::deleted)
-        .filter(|event| future::ready(event.id == channel.id))
-        .next()
-        .expect_some("a deleted channel event will be delivered")
-        .await;
-}
-
-#[tokio::test]
-async fn previously_expired() {
-    // Set up the environment
-
-    let app = fixtures::scratch_app().await;
-    let channel = fixtures::channel::create(&app, &fixtures::ancient()).await;
-    let resume_point = fixtures::boot::resume_point(&app).await;
-
-    // Expire channels
-
-    app.channels()
-        .expire(&fixtures::now())
-        .await
-        .expect("expiring channels always succeeds");
-
-    // Subscribe
-
-    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
-    let super::Response(events) = super::handler(
-        State(app.clone()),
-        subscriber,
-        None,
-        Query(super::QueryParams { resume_point }),
-    )
-    .await
-    .expect("subscribe never fails");
-
-    // Check for expiry event
-    let _ = events
-        .filter_map(fixtures::event::stream::channel)
-        .filter_map(fixtures::event::stream::channel::deleted)
-        .filter(|event| future::ready(event.id == channel.id))
-        .next()
-        .expect_some("a deleted channel event will be delivered")
-        .await;
-}
-
-#[tokio::test]
-async fn deleting() {
-    // Set up the environment
-
-    let app = fixtures::scratch_app().await;
-    let channel = fixtures::channel::create(&app, &fixtures::now()).await;
-    let resume_point = fixtures::boot::resume_point(&app).await;
-
-    // Subscribe
-
-    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
-    let super::Response(events) = super::handler(
-        State(app.clone()),
-        subscriber,
-        None,
-        Query(super::QueryParams { resume_point }),
-    )
-    .await
-    .expect("subscribe never fails");
-
-    // Delete the channel
-
-    app.channels()
-        .delete(&channel.id, &fixtures::now())
-        .await
-        .expect("deleting a valid channel succeeds");
-
-    // Check for delete event
-    let _ = events
-        .filter_map(fixtures::event::stream::channel)
-        .filter_map(fixtures::event::stream::channel::deleted)
-        .filter(|event| future::ready(event.id == channel.id))
-        .next()
-        .expect_some("a deleted channel event will be delivered")
-        .await;
-}
-
-#[tokio::test]
-async fn previously_deleted() {
-    // Set up the environment
-
-    let app = fixtures::scratch_app().await;
-    let channel = fixtures::channel::create(&app, &fixtures::now()).await;
-    let resume_point = fixtures::boot::resume_point(&app).await;
-
-    // Delete the channel
-
-    app.channels()
-        .delete(&channel.id, &fixtures::now())
-        .await
-        .expect("deleting a valid channel succeeds");
-
-    // Subscribe
-
-    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
-    let super::Response(events) = super::handler(
-        State(app.clone()),
-        subscriber,
-        None,
-        Query(super::QueryParams { resume_point }),
-    )
-    .await
-    .expect("subscribe never fails");
-
-    // Check for expiry event
-    let _ = events
-        .filter_map(fixtures::event::stream::channel)
-        .filter_map(fixtures::event::stream::channel::deleted)
-        .filter(|event| future::ready(event.id == channel.id))
-        .next()
-        .expect_some("a deleted channel event will be delivered")
-        .await;
-}
-
-#[tokio::test]
-async fn previously_purged() {
-    // Set up the environment
-
-    let app = fixtures::scratch_app().await;
-    let channel = fixtures::channel::create(&app, &fixtures::ancient()).await;
-    let resume_point = fixtures::boot::resume_point(&app).await;
-
-    // Delete and purge the channel
-
-    app.channels()
-        .delete(&channel.id, &fixtures::ancient())
-        .await
-        .expect("deleting a valid channel succeeds");
-
-    app.channels()
-        .purge(&fixtures::now())
-        .await
-        .expect("purging channels always succeeds");
-
-    // Subscribe
-
-    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
-    let super::Response(events) = super::handler(
-        State(app.clone()),
-        subscriber,
-        None,
-        Query(super::QueryParams { resume_point }),
-    )
-    .await
-    .expect("subscribe never fails");
-
-    // Check for expiry event
-    events
-        .filter_map(fixtures::event::stream::channel)
-        .filter_map(fixtures::event::stream::channel::deleted)
-        .filter(|event| future::ready(event.id == channel.id))
-        .next()
-        .expect_wait("deleted channel events not delivered")
-        .await;
-}
diff --git a/src/event/handlers/stream/test/conversation.rs b/src/event/handlers/stream/test/conversation.rs
new file mode 100644
index 0000000..5e08075
--- /dev/null
+++ b/src/event/handlers/stream/test/conversation.rs
@@ -0,0 +1,273 @@
+use axum::extract::State;
+use axum_extra::extract::Query;
+use futures::{future, stream::StreamExt as _};
+
+use crate::test::fixtures::{self, future::Expect as _};
+
+#[tokio::test]
+async fn creating() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let resume_point = fixtures::boot::resume_point(&app).await;
+
+    // Subscribe
+
+    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
+    let super::Response(events) = super::handler(
+        State(app.clone()),
+        subscriber,
+        None,
+        Query(super::QueryParams { resume_point }),
+    )
+    .await
+    .expect("subscribe never fails");
+
+    // Create a conversation
+
+    let name = fixtures::conversation::propose();
+    let conversation = app
+        .conversations()
+        .create(&name, &fixtures::now())
+        .await
+        .expect("creating a conversation succeeds");
+
+    // Verify conversation created event
+
+    events
+        .filter_map(fixtures::event::stream::conversation)
+        .filter_map(fixtures::event::stream::conversation::created)
+        .filter(|event| future::ready(event.conversation == conversation))
+        .next()
+        .expect_some("conversation created event is delivered")
+        .await;
+}
+
+#[tokio::test]
+async fn previously_created() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let resume_point = fixtures::boot::resume_point(&app).await;
+
+    // Create a conversation
+
+    let name = fixtures::conversation::propose();
+    let conversation = app
+        .conversations()
+        .create(&name, &fixtures::now())
+        .await
+        .expect("creating a conversation succeeds");
+
+    // Subscribe
+
+    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
+    let super::Response(events) = super::handler(
+        State(app.clone()),
+        subscriber,
+        None,
+        Query(super::QueryParams { resume_point }),
+    )
+    .await
+    .expect("subscribe never fails");
+
+    // Verify conversation created event
+
+    let _ = events
+        .filter_map(fixtures::event::stream::conversation)
+        .filter_map(fixtures::event::stream::conversation::created)
+        .filter(|event| future::ready(event.conversation == conversation))
+        .next()
+        .expect_some("conversation created event is delivered")
+        .await;
+}
+
+#[tokio::test]
+async fn expiring() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let conversation = fixtures::conversation::create(&app, &fixtures::ancient()).await;
+    let resume_point = fixtures::boot::resume_point(&app).await;
+
+    // Subscribe
+
+    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
+    let super::Response(events) = super::handler(
+        State(app.clone()),
+        subscriber,
+        None,
+        Query(super::QueryParams { resume_point }),
+    )
+    .await
+    .expect("subscribe never fails");
+
+    // Expire conversations
+
+    app.conversations()
+        .expire(&fixtures::now())
+        .await
+        .expect("expiring conversations always succeeds");
+
+    // Check for expiry event
+    let _ = events
+        .filter_map(fixtures::event::stream::conversation)
+        .filter_map(fixtures::event::stream::conversation::deleted)
+        .filter(|event| future::ready(event.id == conversation.id))
+        .next()
+        .expect_some("a deleted conversation event will be delivered")
+        .await;
+}
+
+#[tokio::test]
+async fn previously_expired() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let conversation = fixtures::conversation::create(&app, &fixtures::ancient()).await;
+    let resume_point = fixtures::boot::resume_point(&app).await;
+
+    // Expire conversations
+
+    app.conversations()
+        .expire(&fixtures::now())
+        .await
+        .expect("expiring conversation always succeeds");
+
+    // Subscribe
+
+    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
+    let super::Response(events) = super::handler(
+        State(app.clone()),
+        subscriber,
+        None,
+        Query(super::QueryParams { resume_point }),
+    )
+    .await
+    .expect("subscribe never fails");
+
+    // Check for expiry event
+    let _ = events
+        .filter_map(fixtures::event::stream::conversation)
+        .filter_map(fixtures::event::stream::conversation::deleted)
+        .filter(|event| future::ready(event.id == conversation.id))
+        .next()
+        .expect_some("a deleted conversation event will be delivered")
+        .await;
+}
+
+#[tokio::test]
+async fn deleting() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let conversation = fixtures::conversation::create(&app, &fixtures::now()).await;
+    let resume_point = fixtures::boot::resume_point(&app).await;
+
+    // Subscribe
+
+    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
+    let super::Response(events) = super::handler(
+        State(app.clone()),
+        subscriber,
+        None,
+        Query(super::QueryParams { resume_point }),
+    )
+    .await
+    .expect("subscribe never fails");
+
+    // Delete the conversation
+
+    app.conversations()
+        .delete(&conversation.id, &fixtures::now())
+        .await
+        .expect("deleting a valid conversation succeeds");
+
+    // Check for delete event
+    let _ = events
+        .filter_map(fixtures::event::stream::conversation)
+        .filter_map(fixtures::event::stream::conversation::deleted)
+        .filter(|event| future::ready(event.id == conversation.id))
+        .next()
+        .expect_some("a deleted conversation event will be delivered")
+        .await;
+}
+
+#[tokio::test]
+async fn previously_deleted() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let conversation = fixtures::conversation::create(&app, &fixtures::now()).await;
+    let resume_point = fixtures::boot::resume_point(&app).await;
+
+    // Delete the conversation
+
+    app.conversations()
+        .delete(&conversation.id, &fixtures::now())
+        .await
+        .expect("deleting a valid conversation succeeds");
+
+    // Subscribe
+
+    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
+    let super::Response(events) = super::handler(
+        State(app.clone()),
+        subscriber,
+        None,
+        Query(super::QueryParams { resume_point }),
+    )
+    .await
+    .expect("subscribe never fails");
+
+    // Check for expiry event
+    let _ = events
+        .filter_map(fixtures::event::stream::conversation)
+        .filter_map(fixtures::event::stream::conversation::deleted)
+        .filter(|event| future::ready(event.id == conversation.id))
+        .next()
+        .expect_some("a deleted conversation event will be delivered")
+        .await;
+}
+
+#[tokio::test]
+async fn previously_purged() {
+    // Set up the environment
+
+    let app = fixtures::scratch_app().await;
+    let conversation = fixtures::conversation::create(&app, &fixtures::ancient()).await;
+    let resume_point = fixtures::boot::resume_point(&app).await;
+
+    // Delete and purge the conversation
+
+    app.conversations()
+        .delete(&conversation.id, &fixtures::ancient())
+        .await
+        .expect("deleting a valid conversation succeeds");
+
+    app.conversations()
+        .purge(&fixtures::now())
+        .await
+        .expect("purging conversations always succeeds");
+
+    // Subscribe
+
+    let subscriber = fixtures::identity::create(&app, &fixtures::now()).await;
+    let super::Response(events) = super::handler(
+        State(app.clone()),
+        subscriber,
+        None,
+        Query(super::QueryParams { resume_point }),
+    )
+    .await
+    .expect("subscribe never fails");
+
+    // Check for expiry event
+    events
+        .filter_map(fixtures::event::stream::conversation)
+        .filter_map(fixtures::event::stream::conversation::deleted)
+        .filter(|event| future::ready(event.id == conversation.id))
+        .next()
+        .expect_wait("deleted conversation events not delivered")
+        .await;
+}
diff --git a/src/event/handlers/stream/test/message.rs b/src/event/handlers/stream/test/message.rs
index 4369996..3fba317 100644
--- a/src/event/handlers/stream/test/message.rs
+++ b/src/event/handlers/stream/test/message.rs
@@ -12,7 +12,7 @@ async fn sending() {
     // Set up the environment
 
     let app = fixtures::scratch_app().await;
-    let channel = fixtures::channel::create(&app, &fixtures::now()).await;
+    let conversation = fixtures::conversation::create(&app, &fixtures::now()).await;
     let resume_point = fixtures::boot::resume_point(&app).await;
 
     // Call the endpoint
@@ -33,7 +33,7 @@ async fn sending() {
     let message = app
         .messages()
         .send(
-            &channel.id,
+            &conversation.id,
             &sender,
             &fixtures::now(),
             &fixtures::message::propose(),
@@ -57,7 +57,7 @@ async fn previously_sent() {
     // Set up the environment
 
     let app = fixtures::scratch_app().await;
-    let channel = fixtures::channel::create(&app, &fixtures::now()).await;
+    let conversation = fixtures::conversation::create(&app, &fixtures::now()).await;
     let resume_point = fixtures::boot::resume_point(&app).await;
 
     // Send a message
@@ -66,7 +66,7 @@ async fn previously_sent() {
     let message = app
         .messages()
         .send(
-            &channel.id,
+            &conversation.id,
             &sender,
             &fixtures::now(),
             &fixtures::message::propose(),
@@ -98,27 +98,30 @@ async fn previously_sent() {
 }
 
 #[tokio::test]
-async fn sent_in_multiple_channels() {
+async fn sent_in_multiple_conversations() {
     // Set up the environment
 
     let app = fixtures::scratch_app().await;
     let sender = fixtures::user::create(&app, &fixtures::now()).await;
     let resume_point = fixtures::boot::resume_point(&app).await;
 
-    let channels = [
-        fixtures::channel::create(&app, &fixtures::now()).await,
-        fixtures::channel::create(&app, &fixtures::now()).await,
+    let conversations = [
+        fixtures::conversation::create(&app, &fixtures::now()).await,
+        fixtures::conversation::create(&app, &fixtures::now()).await,
     ];
 
-    let messages = stream::iter(channels)
-        .then(|channel| {
-            let app = app.clone();
-            let sender = sender.clone();
-            let channel = channel.clone();
-            async move { fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await }
-        })
-        .collect::<Vec<_>>()
-        .await;
+    let messages =
+        stream::iter(conversations)
+            .then(|conversation| {
+                let app = app.clone();
+                let sender = sender.clone();
+                let conversation = conversation.clone();
+                async move {
+                    fixtures::message::send(&app, &conversation, &sender, &fixtures::now()).await
+                }
+            })
+            .collect::<Vec<_>>()
+            .await;
 
     // Call the endpoint
 
@@ -152,14 +155,14 @@ async fn sent_sequentially() {
     // Set up the environment
 
     let app = fixtures::scratch_app().await;
-    let channel = fixtures::channel::create(&app, &fixtures::now()).await;
+    let conversation = fixtures::conversation::create(&app, &fixtures::now()).await;
     let sender = fixtures::user::create(&app, &fixtures::now()).await;
     let resume_point = fixtures::boot::resume_point(&app).await;
 
     let messages = vec![
-        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
-        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
-        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
+        fixtures::message::send(&app, &conversation, &sender, &fixtures::now()).await,
+        fixtures::message::send(&app, &conversation, &sender, &fixtures::now()).await,
+        fixtures::message::send(&app, &conversation, &sender, &fixtures::now()).await,
     ];
 
     // Subscribe
@@ -196,9 +199,9 @@ async fn expiring() {
     // Set up the environment
 
     let app = fixtures::scratch_app().await;
-    let channel = fixtures::channel::create(&app, &fixtures::ancient()).await;
+    let conversation = fixtures::conversation::create(&app, &fixtures::ancient()).await;
     let sender = fixtures::user::create(&app, &fixtures::ancient()).await;
-    let message = fixtures::message::send(&app, &channel, &sender, &fixtures::ancient()).await;
+    let message = fixtures::message::send(&app, &conversation, &sender, &fixtures::ancient()).await;
     let resume_point = fixtures::boot::resume_point(&app).await;
 
     // Subscribe
@@ -235,9 +238,9 @@ async fn previously_expired() {
     // Set up the environment
 
     let app = fixtures::scratch_app().await;
-    let channel = fixtures::channel::create(&app, &fixtures::ancient()).await;
+    let conversation = fixtures::conversation::create(&app, &fixtures::ancient()).await;
     let sender = fixtures::user::create(&app, &fixtures::ancient()).await;
-    let message = fixtures::message::send(&app, &channel, &sender, &fixtures::ancient()).await;
+    let message = fixtures::message::send(&app, &conversation, &sender, &fixtures::ancient()).await;
     let resume_point = fixtures::boot::resume_point(&app).await;
 
     // Expire messages
@@ -274,9 +277,9 @@ async fn deleting() {
     // Set up the environment
 
     let app = fixtures::scratch_app().await;
-    let channel = fixtures::channel::create(&app, &fixtures::now()).await;
+    let conversation = fixtures::conversation::create(&app, &fixtures::now()).await;
     let sender = fixtures::user::create(&app, &fixtures::now()).await;
-    let message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await;
+    let message = fixtures::message::send(&app, &conversation, &sender, &fixtures::now()).await;
     let resume_point = fixtures::boot::resume_point(&app).await;
 
     // Subscribe
@@ -313,9 +316,9 @@ async fn previously_deleted() {
     // Set up the environment
 
     let app = fixtures::scratch_app().await;
-    let channel = fixtures::channel::create(&app, &fixtures::now()).await;
+    let conversation = fixtures::conversation::create(&app, &fixtures::now()).await;
     let sender = fixtures::user::create(&app, &fixtures::now()).await;
-    let message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await;
+    let message = fixtures::message::send(&app, &conversation, &sender, &fixtures::now()).await;
     let resume_point = fixtures::boot::resume_point(&app).await;
 
     // Delete the message
@@ -352,9 +355,9 @@ async fn previously_purged() {
     // Set up the environment
 
     let app = fixtures::scratch_app().await;
-    let channel = fixtures::channel::create(&app, &fixtures::ancient()).await;
+    let conversation = fixtures::conversation::create(&app, &fixtures::ancient()).await;
     let sender = fixtures::user::create(&app, &fixtures::ancient()).await;
-    let message = fixtures::message::send(&app, &channel, &sender, &fixtures::ancient()).await;
+    let message = fixtures::message::send(&app, &conversation, &sender, &fixtures::ancient()).await;
     let resume_point = fixtures::boot::resume_point(&app).await;
 
     // Purge the message
diff --git a/src/event/handlers/stream/test/mod.rs b/src/event/handlers/stream/test/mod.rs
index df43deb..3bc634f 100644
--- a/src/event/handlers/stream/test/mod.rs
+++ b/src/event/handlers/stream/test/mod.rs
@@ -1,4 +1,4 @@
-mod channel;
+mod conversation;
 mod invite;
 mod message;
 mod resume;
diff --git a/src/event/handlers/stream/test/resume.rs b/src/event/handlers/stream/test/resume.rs
index 835d350..a0da692 100644
--- a/src/event/handlers/stream/test/resume.rs
+++ b/src/event/handlers/stream/test/resume.rs
@@ -14,15 +14,16 @@ async fn resume() {
     // Set up the environment
 
     let app = fixtures::scratch_app().await;
-    let channel = fixtures::channel::create(&app, &fixtures::now()).await;
+    let conversation = fixtures::conversation::create(&app, &fixtures::now()).await;
     let sender = fixtures::user::create(&app, &fixtures::now()).await;
     let resume_point = fixtures::boot::resume_point(&app).await;
 
-    let initial_message = fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await;
+    let initial_message =
+        fixtures::message::send(&app, &conversation, &sender, &fixtures::now()).await;
 
     let later_messages = vec![
-        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
-        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
+        fixtures::message::send(&app, &conversation, &sender, &fixtures::now()).await,
+        fixtures::message::send(&app, &conversation, &sender, &fixtures::now()).await,
     ];
 
     // Call the endpoint
@@ -75,8 +76,8 @@ async fn resume() {
 
 // This test verifies a real bug I hit developing the vector-of-sequences
 // approach to resuming events. A small omission caused the event IDs in a
-// resumed stream to _omit_ channels that were in the original stream until
-// those channels also appeared in the resumed stream.
+// resumed stream to _omit_ conversations that were in the original stream
+// until those conversations also appeared in the resumed stream.
 //
 // Clients would see something like
 // * In the original stream, Cfoo=5,Cbar=8
@@ -84,8 +85,8 @@ async fn resume() {
 //
 // Disconnecting and reconnecting a second time, using event IDs from that
 // initial period of the first resume attempt, would then cause the second
-// resume attempt to restart all other channels from the beginning, and not
-// from where the first disconnection happened.
+// resume attempt to restart all other conversations from the beginning, and
+// not from where the first disconnection happened.
 //
 // As we have switched to a single global event sequence number, this scenario
 // can no longer arise, but this test is preserved because the actual behaviour
@@ -97,8 +98,8 @@ async fn serial_resume() {
 
     let app = fixtures::scratch_app().await;
     let sender = fixtures::user::create(&app, &fixtures::now()).await;
-    let channel_a = fixtures::channel::create(&app, &fixtures::now()).await;
-    let channel_b = fixtures::channel::create(&app, &fixtures::now()).await;
+    let conversation_a = fixtures::conversation::create(&app, &fixtures::now()).await;
+    let conversation_b = fixtures::conversation::create(&app, &fixtures::now()).await;
     let resume_point = fixtures::boot::resume_point(&app).await;
 
     // Call the endpoint
@@ -107,8 +108,8 @@ async fn serial_resume() {
 
     let resume_at = {
         let initial_messages = [
-            fixtures::message::send(&app, &channel_a, &sender, &fixtures::now()).await,
-            fixtures::message::send(&app, &channel_b, &sender, &fixtures::now()).await,
+            fixtures::message::send(&app, &conversation_a, &sender, &fixtures::now()).await,
+            fixtures::message::send(&app, &conversation_b, &sender, &fixtures::now()).await,
         ];
 
         // First subscription
@@ -148,11 +149,11 @@ async fn serial_resume() {
     // Resume after disconnect
     let resume_at = {
         let resume_messages = [
-            // Note that channel_b does not appear here. The buggy behaviour
-            // would be masked if channel_b happened to send a new message
+            // Note that conversation_b does not appear here. The buggy behaviour
+            // would be masked if conversation_b happened to send a new message
             // into the resumed event stream.
-            fixtures::message::send(&app, &channel_a, &sender, &fixtures::now()).await,
-            fixtures::message::send(&app, &channel_a, &sender, &fixtures::now()).await,
+            fixtures::message::send(&app, &conversation_a, &sender, &fixtures::now()).await,
+            fixtures::message::send(&app, &conversation_a, &sender, &fixtures::now()).await,
         ];
 
         // Second subscription
@@ -190,12 +191,12 @@ async fn serial_resume() {
 
     // Resume after disconnect a second time
     {
-        // At this point, we can send on either channel and demonstrate the
-        // problem. The resume point should before both of these messages, but
-        // after _all_ prior messages.
+        // At this point, we can send on either conversation and demonstrate
+        // the problem. The resume point should before both of these messages,
+        // but after _all_ prior messages.
         let final_messages = [
-            fixtures::message::send(&app, &channel_a, &sender, &fixtures::now()).await,
-            fixtures::message::send(&app, &channel_b, &sender, &fixtures::now()).await,
+            fixtures::message::send(&app, &conversation_a, &sender, &fixtures::now()).await,
+            fixtures::message::send(&app, &conversation_b, &sender, &fixtures::now()).await,
         ];
 
         // Third subscription
diff --git a/src/event/handlers/stream/test/token.rs b/src/event/handlers/stream/test/token.rs
index e32b489..5af07a0 100644
--- a/src/event/handlers/stream/test/token.rs
+++ b/src/event/handlers/stream/test/token.rs
@@ -9,7 +9,7 @@ async fn terminates_on_token_expiry() {
     // Set up the environment
 
     let app = fixtures::scratch_app().await;
-    let channel = fixtures::channel::create(&app, &fixtures::now()).await;
+    let conversation = fixtures::conversation::create(&app, &fixtures::now()).await;
     let sender = fixtures::user::create(&app, &fixtures::now()).await;
     let resume_point = fixtures::boot::resume_point(&app).await;
 
@@ -37,9 +37,9 @@ async fn terminates_on_token_expiry() {
 
     // These should not be delivered.
     let messages = [
-        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
-        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
-        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
+        fixtures::message::send(&app, &conversation, &sender, &fixtures::now()).await,
+        fixtures::message::send(&app, &conversation, &sender, &fixtures::now()).await,
+        fixtures::message::send(&app, &conversation, &sender, &fixtures::now()).await,
     ];
 
     events
@@ -56,7 +56,7 @@ async fn terminates_on_logout() {
     // Set up the environment
 
     let app = fixtures::scratch_app().await;
-    let channel = fixtures::channel::create(&app, &fixtures::now()).await;
+    let conversation = fixtures::conversation::create(&app, &fixtures::now()).await;
     let sender = fixtures::user::create(&app, &fixtures::now()).await;
     let resume_point = fixtures::boot::resume_point(&app).await;
 
@@ -83,9 +83,9 @@ async fn terminates_on_logout() {
     // These should not be delivered.
 
     let messages = [
-        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
-        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
-        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
+        fixtures::message::send(&app, &conversation, &sender, &fixtures::now()).await,
+        fixtures::message::send(&app, &conversation, &sender, &fixtures::now()).await,
+        fixtures::message::send(&app, &conversation, &sender, &fixtures::now()).await,
     ];
 
     events
@@ -102,7 +102,7 @@ async fn terminates_on_password_change() {
     // Set up the environment
 
     let app = fixtures::scratch_app().await;
-    let channel = fixtures::channel::create(&app, &fixtures::now()).await;
+    let conversation = fixtures::conversation::create(&app, &fixtures::now()).await;
     let sender = fixtures::user::create(&app, &fixtures::now()).await;
     let resume_point = fixtures::boot::resume_point(&app).await;
 
@@ -133,9 +133,9 @@ async fn terminates_on_password_change() {
     // These should not be delivered.
 
     let messages = [
-        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
-        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
-        fixtures::message::send(&app, &channel, &sender, &fixtures::now()).await,
+        fixtures::message::send(&app, &conversation, &sender, &fixtures::now()).await,
+        fixtures::message::send(&app, &conversation, &sender, &fixtures::now()).await,
+        fixtures::message::send(&app, &conversation, &sender, &fixtures::now()).await,
     ];
 
     events
-- 
cgit v1.2.3