opti: Select all collab members (#215)

* feat: use af member stream for populating policies

* fix: add missing sqlx files

* chore: removed unused function

.sqlx/query-0517066279a74e9af59645b8bb152273aa959b610ca3d5e4f8eaecf4f7785dc2.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "\n      SELECT w.* \n      FROM af_workspace w\n      JOIN af_workspace_member wm ON w.workspace_id = wm.workspace_id\n      WHERE wm.uid = (\n         SELECT uid FROM public.af_user WHERE uuid = $1\n      );\n    ",
+  "query": "\n      SELECT w.*\n      FROM af_workspace w\n      JOIN af_workspace_member wm ON w.workspace_id = wm.workspace_id\n      WHERE wm.uid = (\n         SELECT uid FROM public.af_user WHERE uuid = $1\n      );\n    ",
   "describe": {
     "columns": [
       {
@@ -54,5 +54,5 @@
       true
     ]
   },
-  "hash": "5855567a45f990e03f9249e0c591ebb8c9949a9b7e5bec114535c99f89a2a1dd"
+  "hash": "0517066279a74e9af59645b8bb152273aa959b610ca3d5e4f8eaecf4f7785dc2"
 }

.sqlx/query-1141f880140df9d91b1464fde0dcad2e39a10e57e828b0ad44a00621d063d18c.json

@@ -1,20 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "SELECT DISTINCT af_workspace_member.workspace_id FROM af_workspace_member",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "workspace_id",
-        "type_info": "Uuid"
-      }
-    ],
-    "parameters": {
-      "Left": []
-    },
-    "nullable": [
-      false
-    ]
-  },
-  "hash": "1141f880140df9d91b1464fde0dcad2e39a10e57e828b0ad44a00621d063d18c"
-}

.sqlx/query-273cb2255ae88a24dde5e35228a6cfdb5b353e52120b0fa7addf72e539f97344.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "\n    SELECT af_user.uid, af_user.name, af_user.email, af_workspace_member.role_id AS role\n    FROM public.af_workspace_member\n      JOIN public.af_user ON af_workspace_member.uid = af_user.uid\n    WHERE af_workspace_member.workspace_id = $1 \n    AND af_workspace_member.uid = $2 \n    ",
+  "query": "\n    SELECT af_user.uid, af_user.name, af_user.email, af_workspace_member.role_id AS role\n    FROM public.af_workspace_member\n      JOIN public.af_user ON af_workspace_member.uid = af_user.uid\n    WHERE af_workspace_member.workspace_id = $1\n    AND af_workspace_member.uid = $2\n    ",
   "describe": {
     "columns": [
       {
@@ -37,5 +37,5 @@
       false
     ]
   },
-  "hash": "75f37fa6492019be868ea7f8ffea64a3809c1232c548608c375b084440f672f1"
+  "hash": "273cb2255ae88a24dde5e35228a6cfdb5b353e52120b0fa7addf72e539f97344"
 }

.sqlx/query-2b0754f55889a20c294d2a77ba8d3fa34c8174856abfdede34797851183a177a.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n  SELECT EXISTS (\n    SELECT 1\n    FROM public.af_workspace\n    WHERE\n        workspace_id = $1\n        AND owner_uid = (\n            SELECT uid FROM public.af_user WHERE email = $2\n        )\n   ) AS \"is_owner\";\n  ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "is_owner",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "2b0754f55889a20c294d2a77ba8d3fa34c8174856abfdede34797851183a177a"
+}

.sqlx/query-354166a6fa147dc6e17bfc14cb68d3a72a2e7c3aa2d115686deb12086786e034.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "\n     SELECT role_id FROM af_workspace_member\n     WHERE workspace_id = $1 AND uid = $2 \n    ",
+  "query": "\n     SELECT role_id FROM af_workspace_member\n     WHERE workspace_id = $1 AND uid = $2\n    ",
   "describe": {
     "columns": [
       {
@@ -19,5 +19,5 @@
       false
     ]
   },
-  "hash": "fbe68f6ca82fba7184181cdb46546633e9eca1c98842b55aa8fb990aa5a43713"
+  "hash": "354166a6fa147dc6e17bfc14cb68d3a72a2e7c3aa2d115686deb12086786e034"
 }

.sqlx/query-6380f5a6ded2dab8f18de42541c9d77c2f3af512e3f66e1b731ca7c00c9ea8f8.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "\n      INSERT INTO public.af_workspace_member (workspace_id, uid, role_id)\n      SELECT $1, af_user.uid, $3\n      FROM public.af_user \n      WHERE \n        af_user.email = $2 \n      ON CONFLICT (workspace_id, uid)\n      DO NOTHING;\n    ",
+  "query": "\n      INSERT INTO public.af_workspace_member (workspace_id, uid, role_id)\n      SELECT $1, af_user.uid, $3\n      FROM public.af_user\n      WHERE\n        af_user.email = $2\n      ON CONFLICT (workspace_id, uid)\n      DO NOTHING;\n    ",
   "describe": {
     "columns": [],
     "parameters": {
@@ -12,5 +12,5 @@
     },
     "nullable": []
   },
-  "hash": "3f5bbb0546a96584851ed6523dbb85e4d95f0aaf3cb391191d6f7726d56aa20e"
+  "hash": "6380f5a6ded2dab8f18de42541c9d77c2f3af512e3f66e1b731ca7c00c9ea8f8"
 }

.sqlx/query-69ab5d0cc26ae1830849de97d53731dbddbf4189cca04706928848e932b7a72c.json

@@ -1,15 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "\n    DELETE FROM public.af_workspace_member\n    WHERE \n    workspace_id = $1 \n    AND uid = (\n        SELECT uid FROM public.af_user WHERE email = $2\n    )\n    -- Ensure the user to be deleted is not the original owner. \n    -- 1. TODO(nathan): User must transfer ownership to another user first.\n    -- 2. User must have at least one workspace\n    AND uid <> (\n        SELECT owner_uid FROM public.af_workspace WHERE workspace_id = $1\n    );\n    ",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Uuid",
-        "Text"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "69ab5d0cc26ae1830849de97d53731dbddbf4189cca04706928848e932b7a72c"
-}

.sqlx/query-8069f4f5e77baf980972726cfed7322cf743927c68bcfd364dcf054f8b940c92.json

@@ -1,16 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "\n        UPDATE af_workspace_member\n        SET \n            role_id = $1 \n        WHERE workspace_id = $2 AND uid = (\n            SELECT uid FROM af_user WHERE email = $3\n        )\n        ",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Int4",
-        "Uuid",
-        "Text"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "8069f4f5e77baf980972726cfed7322cf743927c68bcfd364dcf054f8b940c92"
-}

.sqlx/query-936faba4e3c8fc3685d68f561a2c2d4f386c77cffde6f25702c19758a12669ce.json

@@ -0,0 +1,16 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        UPDATE af_workspace_member\n        SET\n            role_id = $1\n        WHERE workspace_id = $2 AND uid = (\n            SELECT uid FROM af_user WHERE email = $3\n        )\n        ",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Int4",
+        "Uuid",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "936faba4e3c8fc3685d68f561a2c2d4f386c77cffde6f25702c19758a12669ce"
+}

.sqlx/query-a6b4725375e940cacdcc622b5288a38253db8bd48ba0774344d20fb6957f9ca6.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "\n    WITH workspace_check AS (\n        SELECT EXISTS(\n            SELECT 1\n            FROM af_workspace_member\n            WHERE af_workspace_member.uid = (SELECT uid FROM af_user WHERE uuid = $1) AND\n            af_workspace_member.workspace_id = $3\n        ) AS \"workspace_exists\"\n    ),\n    collab_check AS (\n        SELECT EXISTS(\n            SELECT 1\n            FROM af_collab_member\n            WHERE oid = $2\n        ) AS \"collab_exists\"\n    )\n    SELECT \n        NOT collab_check.collab_exists OR (\n            workspace_check.workspace_exists AND \n            EXISTS(\n                SELECT 1\n                FROM af_collab_member\n                JOIN af_permissions ON af_collab_member.permission_id = af_permissions.id\n                WHERE \n                    af_collab_member.uid = (SELECT uid FROM af_user WHERE uuid = $1) AND \n                    af_collab_member.oid = $2 AND \n                    af_permissions.access_level > 20\n            )\n        ) AS \"permission_check\"\n    FROM workspace_check, collab_check;\n     ",
+  "query": "\n    WITH workspace_check AS (\n        SELECT EXISTS(\n            SELECT 1\n            FROM af_workspace_member\n            WHERE af_workspace_member.uid = (SELECT uid FROM af_user WHERE uuid = $1) AND\n            af_workspace_member.workspace_id = $3\n        ) AS \"workspace_exists\"\n    ),\n    collab_check AS (\n        SELECT EXISTS(\n            SELECT 1\n            FROM af_collab_member\n            WHERE oid = $2\n        ) AS \"collab_exists\"\n    )\n    SELECT\n        NOT collab_check.collab_exists OR (\n            workspace_check.workspace_exists AND\n            EXISTS(\n                SELECT 1\n                FROM af_collab_member\n                JOIN af_permissions ON af_collab_member.permission_id = af_permissions.id\n                WHERE\n                    af_collab_member.uid = (SELECT uid FROM af_user WHERE uuid = $1) AND\n                    af_collab_member.oid = $2 AND\n                    af_permissions.access_level > 20\n            )\n        ) AS \"permission_check\"\n    FROM workspace_check, collab_check;\n     ",
   "describe": {
     "columns": [
       {
@@ -20,5 +20,5 @@
       null
     ]
   },
-  "hash": "83b14b1f8483785ddfeab7752f698b663e78936c40a028492aab63e9dda9dc30"
+  "hash": "a6b4725375e940cacdcc622b5288a38253db8bd48ba0774344d20fb6957f9ca6"
 }

.sqlx/query-ba815f67aab3f302a2982225b72c6113bbd9bc87326e4f0a3b44dadbb5f47920.json

@@ -0,0 +1,32 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT uid, role_id as role, workspace_id FROM af_workspace_member",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "uid",
+        "type_info": "Int8"
+      },
+      {
+        "ordinal": 1,
+        "name": "role",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 2,
+        "name": "workspace_id",
+        "type_info": "Uuid"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "ba815f67aab3f302a2982225b72c6113bbd9bc87326e4f0a3b44dadbb5f47920"
+}

.sqlx/query-c843fb8517b1e364016b85a9e94927673bf8311bfbf723b610d59ecfef3fafce.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n    DELETE FROM public.af_workspace_member\n    WHERE\n    workspace_id = $1\n    AND uid = (\n        SELECT uid FROM public.af_user WHERE email = $2\n    )\n    -- Ensure the user to be deleted is not the original owner.\n    -- 1. TODO(nathan): User must transfer ownership to another user first.\n    -- 2. User must have at least one workspace\n    AND uid <> (\n        SELECT owner_uid FROM public.af_workspace WHERE workspace_id = $1\n    );\n    ",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "c843fb8517b1e364016b85a9e94927673bf8311bfbf723b610d59ecfef3fafce"
+}

.sqlx/query-fa7a89a789e010e29dd4737b5614ebb750e446a4c76e298731b1801dee679278.json

@@ -1,23 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "\n  SELECT EXISTS (\n    SELECT 1 \n    FROM public.af_workspace\n    WHERE \n        workspace_id = $1 \n        AND owner_uid = (\n            SELECT uid FROM public.af_user WHERE email = $2\n        )\n   ) AS \"is_owner\";\n  ",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "is_owner",
-        "type_info": "Bool"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Uuid",
-        "Text"
-      ]
-    },
-    "nullable": [
-      null
-    ]
-  },
-  "hash": "fa7a89a789e010e29dd4737b5614ebb750e446a4c76e298731b1801dee679278"
-}

libs/database-entity/src/pg_row.rs

@@ -47,6 +47,13 @@ pub struct AFUserProfileRow {
   pub latest_workspace_id: Option<Uuid>,
 }
 
+#[derive(FromRow, Serialize, Deserialize)]
+pub struct AFWorkspaceMemberPermRow {
+  pub uid: i64,
+  pub role: AFRole,
+  pub workspace_id: Uuid,
+}
+
 #[derive(FromRow, Serialize, Deserialize)]
 pub struct AFWorkspaceMemberRow {
   pub uid: i64,

libs/database/src/workspace.rs

@@ -1,4 +1,5 @@
-use database_entity::dto::AFRole;
+use database_entity::{dto::AFRole, pg_row::AFWorkspaceMemberPermRow};
+use futures_util::stream::BoxStream;
 use sqlx::{
   types::{uuid, Uuid},
   Executor, PgPool, Postgres, Transaction,
@@ -49,7 +50,7 @@ pub async fn select_user_role<'a, E: Executor<'a, Database = Postgres>>(
   let row = sqlx::query_scalar!(
     r#"
      SELECT role_id FROM af_workspace_member
-     WHERE workspace_id = $1 AND uid = $2 
+     WHERE workspace_id = $1 AND uid = $2
     "#,
     workspace_uuid,
     uid
@@ -88,16 +89,16 @@ pub async fn select_user_can_edit_collab(
             WHERE oid = $2
         ) AS "collab_exists"
     )
-    SELECT 
+    SELECT
         NOT collab_check.collab_exists OR (
-            workspace_check.workspace_exists AND 
+            workspace_check.workspace_exists AND
             EXISTS(
                 SELECT 1
                 FROM af_collab_member
                 JOIN af_permissions ON af_collab_member.permission_id = af_permissions.id
-                WHERE 
-                    af_collab_member.uid = (SELECT uid FROM af_user WHERE uuid = $1) AND 
-                    af_collab_member.oid = $2 AND 
+                WHERE
+                    af_collab_member.uid = (SELECT uid FROM af_user WHERE uuid = $1) AND
+                    af_collab_member.oid = $2 AND
                     af_permissions.access_level > 20
             )
         ) AS "permission_check"
@@ -125,9 +126,9 @@ pub async fn insert_workspace_member_with_txn(
     r#"
       INSERT INTO public.af_workspace_member (workspace_id, uid, role_id)
       SELECT $1, af_user.uid, $3
-      FROM public.af_user 
-      WHERE 
-        af_user.email = $2 
+      FROM public.af_user
+      WHERE
+        af_user.email = $2
       ON CONFLICT (workspace_id, uid)
       DO NOTHING;
     "#,
@@ -165,8 +166,8 @@ pub async fn upsert_workspace_member(
   sqlx::query!(
     r#"
         UPDATE af_workspace_member
-        SET 
-            role_id = $1 
+        SET
+            role_id = $1
         WHERE workspace_id = $2 AND uid = (
             SELECT uid FROM af_user WHERE email = $3
         )
@@ -191,10 +192,10 @@ pub async fn delete_workspace_members(
   let is_owner = sqlx::query_scalar!(
     r#"
   SELECT EXISTS (
-    SELECT 1 
+    SELECT 1
     FROM public.af_workspace
-    WHERE 
-        workspace_id = $1 
+    WHERE
+        workspace_id = $1
         AND owner_uid = (
             SELECT uid FROM public.af_user WHERE email = $2
         )
@@ -216,12 +217,12 @@ pub async fn delete_workspace_members(
   sqlx::query!(
     r#"
     DELETE FROM public.af_workspace_member
-    WHERE 
-    workspace_id = $1 
+    WHERE
+    workspace_id = $1
     AND uid = (
         SELECT uid FROM public.af_user WHERE email = $2
     )
-    -- Ensure the user to be deleted is not the original owner. 
+    -- Ensure the user to be deleted is not the original owner.
     -- 1. TODO(nathan): User must transfer ownership to another user first.
     -- 2. User must have at least one workspace
     AND uid <> (
@@ -237,24 +238,15 @@ pub async fn delete_workspace_members(
   Ok(())
 }
 
-pub async fn select_all_workspace_members(
+pub fn select_workspace_member_perm_stream(
   pg_pool: &PgPool,
-) -> Result<Vec<(String, Vec<AFWorkspaceMemberRow>)>, AppError> {
-  let workspaces: Vec<_> =
-    sqlx::query!("SELECT DISTINCT af_workspace_member.workspace_id FROM af_workspace_member")
-      .fetch_all(pg_pool)
-      .await?
-      .into_iter()
-      .map(|r| r.workspace_id)
-      .collect();
-
-  let mut workspace_members = Vec::with_capacity(workspaces.len());
-  for id in workspaces {
-    let members = select_workspace_member_list(pg_pool, &id).await?;
-    workspace_members.push((id.to_string(), members));
-  }
-
-  Ok(workspace_members)
+) -> Result<BoxStream<'_, sqlx::Result<AFWorkspaceMemberPermRow>>, AppError> {
+  let stream = sqlx::query_as!(
+    AFWorkspaceMemberPermRow,
+    "SELECT uid, role_id as role, workspace_id FROM af_workspace_member"
+  )
+  .fetch(pg_pool);
+  Ok(stream)
 }
 
 /// returns a list of workspace members, sorted by their creation time.
@@ -292,8 +284,8 @@ pub async fn select_workspace_member(
     SELECT af_user.uid, af_user.name, af_user.email, af_workspace_member.role_id AS role
     FROM public.af_workspace_member
       JOIN public.af_user ON af_workspace_member.uid = af_user.uid
-    WHERE af_workspace_member.workspace_id = $1 
-    AND af_workspace_member.uid = $2 
+    WHERE af_workspace_member.workspace_id = $1
+    AND af_workspace_member.uid = $2
     "#,
     workspace_id,
     uid,
@@ -366,7 +358,7 @@ pub async fn select_user_workspace<'a, E: Executor<'a, Database = Postgres>>(
   let workspaces = sqlx::query_as!(
     AFWorkspaceRow,
     r#"
-      SELECT w.* 
+      SELECT w.*
       FROM af_workspace w
       JOIN af_workspace_member wm ON w.workspace_id = wm.workspace_id
       WHERE wm.uid = (

src/biz/casbin/adapter.rs

@@ -6,11 +6,13 @@ use casbin::Filter;
 use casbin::Model;
 use casbin::Result;
 use database::collab::select_all_collab_members;
-use database::workspace::select_all_workspace_members;
+use database::workspace::select_workspace_member_perm_stream;
 use database_entity::dto::AFAccessLevel;
 use database_entity::dto::AFCollabMember;
-use database_entity::pg_row::AFWorkspaceMemberRow;
+use database_entity::pg_row::AFWorkspaceMemberPermRow;
+use futures_util::stream::BoxStream;
 use sqlx::PgPool;
+use tokio_stream::StreamExt;
 
 /// Implmentation of [`casbin::Adapter`] for access control authorisation.
 /// Access control policies that are managed by workspace and collab CRUD.
@@ -41,33 +43,32 @@ fn create_collab_policies(collab_members: Vec<(String, Vec<AFCollabMember>)>) ->
   policies
 }
 
-fn create_workspace_policies(
-  workspace_members: Vec<(String, Vec<AFWorkspaceMemberRow>)>,
-) -> Vec<Vec<String>> {
+async fn create_workspace_policies(
+  mut stream: BoxStream<'_, sqlx::Result<AFWorkspaceMemberPermRow>>,
+) -> Result<Vec<Vec<String>>> {
   let mut policies: Vec<Vec<String>> = Vec::new();
-  for (oid, members) in workspace_members {
-    for m in members {
-      let p = [
-        m.uid.to_string(),
-        ObjectType::Workspace(&oid).to_string(),
-        i32::from(m.role).to_string(),
-      ]
-      .to_vec();
-      policies.push(p);
-    }
+
+  while let Some(result) = stream.next().await {
+    let member_permission = result.map_err(|err| AdapterError(Box::new(err)))?;
+    let policy = [
+      member_permission.uid.to_string(),
+      ObjectType::Workspace(&member_permission.workspace_id.to_string()).to_string(),
+      i32::from(member_permission.role).to_string(),
+    ]
+    .to_vec();
+    policies.push(policy);
   }
 
-  policies
+  Ok(policies)
 }
 
 #[async_trait]
 impl Adapter for PgAdapter {
   async fn load_policy(&mut self, model: &mut dyn Model) -> Result<()> {
-    let workspace_members = select_all_workspace_members(&self.pg_pool)
-      .await
+    let workspace_member_perm_stream = select_workspace_member_perm_stream(&self.pg_pool)
       .map_err(|err| AdapterError(Box::new(err)))?;
+    let workspace_policies = create_workspace_policies(workspace_member_perm_stream).await?;
 
-    let workspace_policies = create_workspace_policies(workspace_members);
     // Policy definition `p` of type `p`. See `model.conf`
     model.add_policies("p", "p", workspace_policies);