Ban

Cargo.toml

@@ -55,6 +55,8 @@ svc-error = { version = "0.4", features = [
     "sentry-extension",
     "sqlx",
 ] }
+svc-events = { git = "https://github.com/foxford/svc-events", branch = "ULMS-1896/ban-events" }
+svc-nats-client = { git = "https://github.com/foxford/svc-nats-client/", branch = "ULMS-1896/add-consumer-move-event-id" }
 svc-utils = { version = "0.6.0", features = ["log-middleware", "metrics-middleware", "cors-middleware", "authn-extractor"] }
 tokio = { version = "1.17", features = ["full"] }
 tower = { version = "0.4", features = [ "timeout" ] }

docker/Dockerfile

@@ -1,7 +1,7 @@
 ## -----------------------------------------------------------------------------
 ## Build
 ## -----------------------------------------------------------------------------
-FROM rust:1.64.0-slim-buster as build-stage
+FROM rust:1.69.0-slim-buster as build-stage
 
 RUN apt update && apt install -y --no-install-recommends \
   pkg-config \

migrations/20230517104549_add_ban_account_op_table.sql

@@ -0,0 +1,8 @@
+CREATE TABLE IF NOT EXISTS ban_account_op (
+    user_account account_id PRIMARY KEY,
+    last_op_id uuid NOT NULL,
+    video_complete boolean NOT NULL DEFAULT false,
+    event_access_complete boolean NOT NULL DEFAULT false
+);
+
+CREATE SEQUENCE IF NOT EXISTS ban_entity_seq_id;

rust-toolchain.toml

@@ -1,3 +1,3 @@
 [toolchain]
-channel = "1.64.0"
+channel = "1.69.0"
 components = ["rustfmt", "clippy"]

sqlx-data.json

@@ -1526,6 +1526,78 @@
     },
     "query": "\n                        SELECT\n                            id::text AS \"id!: String\"\n                        FROM class\n                        WHERE conference_room_id = $1\n                    "
   },
+  "8cdcd79a99763311030cf8cdb85effaf8031104a08a3b118bdaaf33c4269980f": {
+    "describe": {
+      "columns": [
+        {
+          "name": "user_account: _",
+          "ordinal": 0,
+          "type_info": {
+            "Custom": {
+              "kind": {
+                "Composite": [
+                  [
+                    "label",
+                    "Text"
+                  ],
+                  [
+                    "audience",
+                    "Text"
+                  ]
+                ]
+              },
+              "name": "account_id"
+            }
+          }
+        },
+        {
+          "name": "last_op_id: _",
+          "ordinal": 1,
+          "type_info": "Uuid"
+        },
+        {
+          "name": "video_complete",
+          "ordinal": 2,
+          "type_info": "Bool"
+        },
+        {
+          "name": "event_access_complete",
+          "ordinal": 3,
+          "type_info": "Bool"
+        }
+      ],
+      "nullable": [
+        false,
+        false,
+        false,
+        false
+      ],
+      "parameters": {
+        "Left": [
+          "Record"
+        ]
+      }
+    },
+    "query": "\n            SELECT\n                user_account AS \"user_account: _\",\n                last_op_id AS \"last_op_id: _\",\n                video_complete,\n                event_access_complete\n            FROM ban_account_op\n            WHERE\n                user_account = $1\n            LIMIT 1;\n            "
+  },
+  "a4fc28b7fa8a6a5f4051750a2447784a10a221632dbea080fe1d4c980de076fd": {
+    "describe": {
+      "columns": [
+        {
+          "name": "value!: i64",
+          "ordinal": 0,
+          "type_info": "Int8"
+        }
+      ],
+      "nullable": [
+        null
+      ],
+      "parameters": {
+        "Left": []
+      }
+    },
+    "query": "SELECT nextval('ban_entity_seq_id') as \"value!: i64\";"
+  },
   "ac4ac9431173165543dbc459bb3b2b9b4461f7016aa6d6967cc9f5f832f208bb": {
     "describe": {
       "columns": [
@@ -1681,6 +1753,80 @@
     },
     "query": "\n            INSERT INTO recording (class_id, rtc_id, segments, modified_segments, stream_uri, started_at, adjusted_at, transcoded_at, created_by)\n            VALUES ($1, $2, $3, $4, $5, NOW(), NOW(), NOW(), $6)\n            RETURNING\n                id,\n                class_id,\n                rtc_id,\n                stream_uri,\n                segments AS \"segments!: Option<Segments>\",\n                started_at,\n                modified_segments AS \"modified_segments!: Option<Segments>\",\n                created_at,\n                adjusted_at,\n                transcoded_at,\n                created_by AS \"created_by: AgentId\",\n                deleted_at\n            "
   },
+  "ad5a4328c5367ea842463346d51759c429921081fb8c9c8e103970d77209ff10": {
+    "describe": {
+      "columns": [
+        {
+          "name": "user_account: _",
+          "ordinal": 0,
+          "type_info": {
+            "Custom": {
+              "kind": {
+                "Composite": [
+                  [
+                    "label",
+                    "Text"
+                  ],
+                  [
+                    "audience",
+                    "Text"
+                  ]
+                ]
+              },
+              "name": "account_id"
+            }
+          }
+        },
+        {
+          "name": "last_op_id",
+          "ordinal": 1,
+          "type_info": "Uuid"
+        },
+        {
+          "name": "video_complete",
+          "ordinal": 2,
+          "type_info": "Bool"
+        },
+        {
+          "name": "event_access_complete",
+          "ordinal": 3,
+          "type_info": "Bool"
+        }
+      ],
+      "nullable": [
+        false,
+        false,
+        false,
+        false
+      ],
+      "parameters": {
+        "Left": [
+          {
+            "Custom": {
+              "kind": {
+                "Composite": [
+                  [
+                    "label",
+                    "Text"
+                  ],
+                  [
+                    "audience",
+                    "Text"
+                  ]
+                ]
+              },
+              "name": "account_id"
+            }
+          },
+          "Uuid",
+          "Bool",
+          "Bool",
+          "Uuid"
+        ]
+      }
+    },
+    "query": "\n            INSERT INTO ban_account_op (user_account, last_op_id, video_complete, event_access_complete)\n            VALUES ($1, $2, COALESCE($3, false), COALESCE($4, false))\n            ON CONFLICT (user_account) DO UPDATE\n            SET\n                video_complete        = COALESCE(EXCLUDED.video_complete, ban_account_op.video_complete),\n                event_access_complete = COALESCE(EXCLUDED.event_access_complete, ban_account_op.event_access_complete),\n                last_op_id            = EXCLUDED.last_op_id\n            WHERE\n                -- allow to 'complete' operation iff there's no change in last_op_id\n                -- or allow to do upsert without real changes so we can process\n                -- the same message twice\n                ban_account_op.last_op_id = EXCLUDED.last_op_id OR\n                -- allow change op id iff the previous operation is completed\n                (\n                    ban_account_op.last_op_id            = $5 AND\n                    ban_account_op.video_complete        = true AND\n                    ban_account_op.event_access_complete = true\n                )\n            RETURNING\n                user_account AS \"user_account: _\",\n                last_op_id,\n                video_complete,\n                event_access_complete\n            "
+  },
   "c67188dde7672c71f7e14a0ef09047934fbf808e5541e1b35c88004f36c16c8b": {
     "describe": {
       "columns": [

src/app/api/v1/account/ban.rs

@@ -0,0 +1,83 @@
+use std::sync::Arc;
+
+use axum::{extract::Path, response::Response, Extension, Json};
+use hyper::Body;
+use serde_derive::Deserialize;
+use svc_authn::AccountId;
+use uuid::Uuid;
+
+use svc_utils::extractors::AccountIdExtractor;
+
+use crate::{
+    app::{
+        api::v1::find_class,
+        error::{ErrorExt, ErrorKind as AppErrorKind},
+        metrics::AuthorizeMetrics,
+        stage, AppContext, AuthzObject,
+    },
+    db::ban_account_op,
+};
+
+use super::AppResult;
+
+#[derive(Deserialize)]
+pub struct BanPayload {
+    // TODO: maybe Option?
+    last_seen_op_id: Uuid,
+    ban: bool,
+    class_id: Uuid,
+}
+
+pub async fn ban(
+    Extension(ctx): Extension<Arc<dyn AppContext>>,
+    Path(account_to_ban): Path<AccountId>,
+    AccountIdExtractor(account_id): AccountIdExtractor,
+    Json(payload): Json<BanPayload>,
+) -> AppResult {
+    let class = find_class(ctx.as_ref(), payload.class_id)
+        .await
+        .error(AppErrorKind::ClassNotFound)?;
+
+    let object = AuthzObject::new(&["classrooms", &class.id().to_string()]).into();
+    ctx.authz()
+        .authorize(
+            class.audience().to_owned(),
+            account_id.clone(),
+            object,
+            "update".into(),
+        )
+        .await
+        .measure()?;
+
+    let mut conn = ctx
+        .get_conn()
+        .await
+        .error(AppErrorKind::DbConnAcquisitionFailed)?;
+
+    let last_ban_account_op = ban_account_op::ReadQuery::by_id(&account_to_ban)
+        .execute(&mut conn)
+        .await
+        .error(AppErrorKind::DbQueryFailed)?;
+
+    if let Some(op) = last_ban_account_op {
+        if op.last_op_id != payload.last_seen_op_id {
+            return Err(AppErrorKind::OperationIdObsolete.into());
+        }
+
+        if !op.complete() {
+            return Err(AppErrorKind::OperationInProgress.into());
+        }
+    }
+
+    stage::ban_intent::start(
+        ctx.as_ref(),
+        &mut conn,
+        payload.ban,
+        &class,
+        account_to_ban,
+        payload.last_seen_op_id,
+    )
+    .await?;
+
+    Ok(Response::builder().status(200).body(Body::empty()).unwrap())
+}

src/app/api/v1/account.rs → src/app/api/v1/account/mod.rs

@@ -4,6 +4,10 @@ use axum::{extract::Path, Extension};
 use svc_agent::AccountId;
 use svc_utils::extractors::AccountIdExtractor;
 
+mod ban;
+
+pub use ban::ban;
+
 use crate::{
     app::{
         api::IntoJsonResponse,

src/app/error.rs

@@ -40,6 +40,11 @@ pub enum ErrorKind {
     InternalFailure,
     CreationWhiteboardFailed,
     ClassAlreadyEstablished,
+    OperationIdObsolete,
+    OperationInProgress,
+    OperationFailure,
+    NatsPublishFailed,
+    NatsClientNotFound,
 }
 
 impl ErrorKind {
@@ -191,6 +196,36 @@ impl From<ErrorKind> for ErrorKindProperties {
                 title: "Class already established",
                 is_notify_sentry: false,
             },
+            ErrorKind::OperationIdObsolete => ErrorKindProperties {
+                status: ResponseStatus::CONFLICT,
+                kind: "operation_id_obsolete",
+                title: "Operation id obsolete, should fetch latest state",
+                is_notify_sentry: false,
+            },
+            ErrorKind::OperationInProgress => ErrorKindProperties {
+                status: ResponseStatus::CONFLICT,
+                kind: "operation_in_progress",
+                title: "Operation is not completed yet, retry later",
+                is_notify_sentry: false,
+            },
+            ErrorKind::OperationFailure => ErrorKindProperties {
+                status: ResponseStatus::INTERNAL_SERVER_ERROR,
+                kind: "operation_failure",
+                title: "Operation failed really bad",
+                is_notify_sentry: true,
+            },
+            ErrorKind::NatsPublishFailed => ErrorKindProperties {
+                status: ResponseStatus::UNPROCESSABLE_ENTITY,
+                kind: "nats_publish_failed",
+                title: "Nats publish failed",
+                is_notify_sentry: true,
+            },
+            ErrorKind::NatsClientNotFound => ErrorKindProperties {
+                status: ResponseStatus::FAILED_DEPENDENCY,
+                kind: "nats_client_not_found",
+                title: "Nats client not found",
+                is_notify_sentry: true,
+            },
         }
     }
 }

src/app/http.rs

@@ -157,6 +157,7 @@ fn utils_router() -> Router {
             "/api/v1/account/properties/:property_id",
             get(account::read_property).put(account::update_property),
         )
+        .metered_route("/api/v1/account/:id/ban", post(account::ban))
         .metered_route(
             "/api/v1/transcoding/minigroup/:id/restart",
             post(restart_transcoding_minigroup),