Add class_id to room

docs/src/api/room.md

@@ -11,6 +11,7 @@ created_at |        int | _required_ | Room creation timestamp in seconds.
 backend    |     string | _required_ | Room backend, either `janus` or `none`.
 reserve    |        int | _optional_ | The number of slots for agents reserved on the backend.
 tags       |       json | {}         | Arbitrary tags object associated with the room.
+class_id   |       uuid | _optional_ | Dispatcher class identifier which the room belongs to.
 
 
 Room can be unbounded, ie its closing timestamp is null.

migrations/2021-04-15-120829_add_class_id_to_room/down.sql

@@ -0,0 +1 @@
+ALTER TABLE room DROP COLUMN class_id;

migrations/2021-04-15-120829_add_class_id_to_room/up.sql

@@ -0,0 +1 @@
+ALTER TABLE room ADD COLUMN class_id UUID;

src/app/endpoint/room.rs

@@ -51,6 +51,7 @@ pub(crate) struct CreateRequest {
     rtc_sharing_policy: Option<RtcSharingPolicy>,
     reserve: Option<i32>,
     tags: Option<JsonValue>,
+    class_id: Option<Uuid>,
 }
 
 pub(crate) struct CreateHandler;
@@ -90,6 +91,10 @@ impl RequestHandler for CreateHandler {
                 q = q.tags(tags);
             }
 
+            if let Some(class_id) = payload.class_id {
+                q = q.class_id(class_id);
+            }
+
             let conn = context.get_conn()?;
             q.execute(&conn)?
         };
@@ -170,6 +175,7 @@ pub(crate) struct UpdateRequest {
     audience: Option<String>,
     reserve: Option<Option<i32>>,
     tags: Option<JsonValue>,
+    class_id: Option<Uuid>,
 }
 pub(crate) struct UpdateHandler;
 
@@ -245,6 +251,7 @@ impl RequestHandler for UpdateHandler {
                 .audience(payload.audience)
                 .reserve(payload.reserve)
                 .tags(payload.tags)
+                .class_id(payload.class_id)
                 .execute(&conn)?
         };
 
@@ -461,6 +468,7 @@ mod test {
                 // Make room.create request.
                 let mut context = TestContext::new(db.clone(), authz);
                 let time = (Bound::Unbounded, Bound::Unbounded);
+                let class_id = Uuid::new_v4();
 
                 let payload = CreateRequest {
                     time: time.clone(),
@@ -469,6 +477,7 @@ mod test {
                     rtc_sharing_policy: Some(db::rtc::SharingPolicy::Shared),
                     reserve: Some(123),
                     tags: Some(json!({ "foo": "bar" })),
+                    class_id: Some(class_id),
                 };
 
                 let messages = handle_request::<CreateHandler>(&mut context, &agent, payload)
@@ -483,6 +492,7 @@ mod test {
                 assert_eq!(room.rtc_sharing_policy(), db::rtc::SharingPolicy::Shared);
                 assert_eq!(room.reserve(), Some(123));
                 assert_eq!(room.tags(), &json!({ "foo": "bar" }));
+                assert_eq!(room.class_id(), Some(class_id));
 
                 // Assert notification.
                 let (room, evp, topic) = find_event::<Room>(messages.as_slice());
@@ -493,6 +503,7 @@ mod test {
                 assert_eq!(room.rtc_sharing_policy(), db::rtc::SharingPolicy::Shared);
                 assert_eq!(room.reserve(), Some(123));
                 assert_eq!(room.tags(), &json!({ "foo": "bar" }));
+                assert_eq!(room.class_id(), Some(class_id));
             });
         }
 
@@ -510,6 +521,7 @@ mod test {
                     rtc_sharing_policy: Some(db::rtc::SharingPolicy::Shared),
                     reserve: None,
                     tags: None,
+                    class_id: None,
                 };
 
                 let err = handle_request::<CreateHandler>(&mut context, &agent, payload)
@@ -651,6 +663,7 @@ mod test {
 
                 // Make room.update request.
                 let mut context = TestContext::new(db, authz);
+                let class_id = Uuid::new_v4();
 
                 let time = (
                     Bound::Included(now + Duration::minutes(50)),
@@ -663,6 +676,7 @@ mod test {
                     reserve: Some(Some(123)),
                     tags: Some(json!({"foo": "bar"})),
                     audience: None,
+                    class_id: Some(class_id),
                 };
 
                 let messages = handle_request::<UpdateHandler>(&mut context, &agent, payload)
@@ -681,6 +695,7 @@ mod test {
                 );
                 assert_eq!(resp_room.reserve(), Some(123));
                 assert_eq!(resp_room.tags(), &json!({"foo": "bar"}));
+                assert_eq!(resp_room.class_id(), Some(class_id));
             });
         }
 
@@ -727,6 +742,7 @@ mod test {
                     reserve: Some(Some(123)),
                     tags: Some(json!({"foo": "bar"})),
                     audience: None,
+                    class_id: None,
                 };
 
                 handle_request::<UpdateHandler>(&mut context, &agent, payload)
@@ -776,8 +792,7 @@ mod test {
                     id: room.id(),
                     time: Some(time),
                     reserve: Some(Some(123)),
-                    audience: None,
-                    tags: None,
+                    ..Default::default()
                 };
 
                 let messages = handle_request::<UpdateHandler>(&mut context, &agent, payload)
@@ -848,9 +863,7 @@ mod test {
                 let payload = UpdateRequest {
                     id: room.id(),
                     time: Some(time),
-                    reserve: None,
-                    audience: None,
-                    tags: None,
+                    ..Default::default()
                 };
 
                 handle_request::<UpdateHandler>(&mut context, &agent, payload)

src/app/endpoint/system.rs

@@ -106,7 +106,7 @@ impl RequestHandler for VacuumHandler {
                         recording.rtc_id(),
                         &config.backend,
                         &config.bucket,
-                        &record_name(&recording),
+                        &record_name(&recording, &room),
                     ),
                     backend.id(),
                     context.start_timestamp(),
@@ -159,7 +159,7 @@ where
             RecordingStatus::Ready => Some(format!(
                 "s3://{}/{}",
                 &upload_config(context, &room)?.bucket,
-                record_name(&recording)
+                record_name(&recording, &room)
             )),
         };
 
@@ -209,8 +209,19 @@ fn upload_config<'a, C: Context>(
         .error(AppErrorKind::ConfigKeyMissing)
 }
 
-fn record_name(recording: &Recording) -> String {
-    format!("{}.source.webm", recording.rtc_id())
+fn record_name(recording: &Recording, room: &Room) -> String {
+    let prefix = match room.rtc_sharing_policy() {
+        SharingPolicy::Owned => {
+            if let Some(class_id) = room.class_id() {
+                format!("{}/", class_id)
+            } else {
+                String::from("")
+            }
+        }
+        _ => String::from(""),
+    };
+
+    format!("{}{}.source.webm", prefix, recording.rtc_id())
 }
 
 ///////////////////////////////////////////////////////////////////////////////

src/db/room.rs

@@ -30,6 +30,7 @@ type AllColumns = (
     room::tags,
     room::backend_id,
     room::rtc_sharing_policy,
+    room::class_id,
 );
 
 const ALL_COLUMNS: AllColumns = (
@@ -42,6 +43,7 @@ const ALL_COLUMNS: AllColumns = (
     room::tags,
     room::backend_id,
     room::rtc_sharing_policy,
+    room::class_id,
 );
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -102,6 +104,7 @@ pub(crate) struct Object {
     #[serde(skip_serializing_if = "Option::is_none")]
     backend_id: Option<AgentId>,
     rtc_sharing_policy: RtcSharingPolicy,
+    class_id: Option<Uuid>,
 }
 
 impl Object {
@@ -140,6 +143,10 @@ impl Object {
     pub(crate) fn rtc_sharing_policy(&self) -> RtcSharingPolicy {
         self.rtc_sharing_policy
     }
+
+    pub(crate) fn class_id(&self) -> Option<Uuid> {
+        self.class_id
+    }
 }
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -236,6 +243,7 @@ pub(crate) struct InsertQuery<'a> {
     tags: Option<&'a JsonValue>,
     backend_id: Option<&'a AgentId>,
     rtc_sharing_policy: RtcSharingPolicy,
+    class_id: Option<Uuid>,
 }
 
 impl<'a> InsertQuery<'a> {
@@ -248,6 +256,7 @@ impl<'a> InsertQuery<'a> {
             tags: None,
             backend_id: None,
             rtc_sharing_policy,
+            class_id: None,
         }
     }
 
@@ -273,6 +282,13 @@ impl<'a> InsertQuery<'a> {
         }
     }
 
+    pub(crate) fn class_id(self, class_id: Uuid) -> Self {
+        Self {
+            class_id: Some(class_id),
+            ..self
+        }
+    }
+
     pub(crate) fn execute(&self, conn: &PgConnection) -> Result<Object, Error> {
         use crate::schema::room::dsl::room;
         use diesel::RunQueryDsl;
@@ -292,6 +308,7 @@ pub(crate) struct UpdateQuery<'a> {
     reserve: Option<Option<i32>>,
     tags: Option<JsonValue>,
     backend_id: Option<&'a AgentId>,
+    class_id: Option<Uuid>,
 }
 
 impl<'a> UpdateQuery<'a> {
@@ -322,6 +339,10 @@ impl<'a> UpdateQuery<'a> {
         Self { backend_id, ..self }
     }
 
+    pub(crate) fn class_id(self, class_id: Option<Uuid>) -> Self {
+        Self { class_id, ..self }
+    }
+
     pub(crate) fn execute(&self, conn: &PgConnection) -> Result<Object, Error> {
         use diesel::prelude::*;
 

src/schema.rs

@@ -79,6 +79,7 @@ table! {
         tags -> Json,
         backend_id -> Nullable<Agent_id>,
         rtc_sharing_policy -> Rtc_sharing_policy,
+        class_id -> Nullable<Uuid>,
     }
 }