Merge pull request #96 from nightly-labs/more-db-schemas

More db schemas

.github/workflows/connect-test-local.yml

@@ -48,9 +48,12 @@ jobs:
         working-directory: ./infra
         run: |
           ( docker-compose logs --follow  &  ) | grep -q "database system is ready to accept connections"
+      - name: Prepare db tables
+        run: |
+          cargo run --bin tables_migration
       - name: run cargo test
         run: |
-          cargo test
+          cargo test -- --test-threads=1
           cargo run --bin nightly-connect-server &
       - name: test base local
         run: |

database/migrations/000000000003_sessions.sql

@@ -0,0 +1,23 @@
+CREATE TABLE sessions(
+    session_id TEXT NOT NULL UNIQUE,
+    app_id TEXT NOT NULL,
+    app_metadata TEXT NOT NULL,
+    app_ip_address TEXT NOT NULL,
+    persistent BOOLEAN NOT NULL,
+    network TEXT NOT NULL,
+    client_id TEXT,
+    client_device TEXT,
+    client_metadata TEXT,
+    client_notification_endpoint TEXT,
+    client_connected_at BIGINT,
+    session_open_timestamp BIGINT NOT NULL,
+    session_close_timestamp BIGINT
+);
+
+CREATE UNIQUE INDEX sessions_session_id ON sessions(session_id);
+
+ALTER TABLE sessions
+ADD CONSTRAINT fk_sessions_registered_apps
+FOREIGN KEY (app_id)
+REFERENCES registered_apps (app_id)
+ON DELETE CASCADE;

database/migrations/000000000004_request_status.sql

@@ -0,0 +1,8 @@
+CREATE TYPE request_status_enum AS ENUM (
+    'Pending', 
+    'Completed', 
+    'Failed', 
+    'Rejected', 
+    'TimedOut', 
+    'Unknown'
+);

database/migrations/000000000005_requests.sql

@@ -0,0 +1,16 @@
+CREATE TABLE requests(
+    request_id TEXT NOT NULL UNIQUE,
+    request_type TEXT NOT NULL,
+    session_id TEXT NOT NULL,
+    request_status request_status_enum NOT NULL,
+    network TEXT NOT NULL,
+    creation_timestamp BIGINT NOT NULL
+);
+
+CREATE UNIQUE INDEX requests_request_id ON requests(request_id);
+
+ALTER TABLE requests
+ADD CONSTRAINT fk_requests_sessions
+FOREIGN KEY (session_id)
+REFERENCES sessions (session_id)
+ON DELETE CASCADE;

database/src/bin/tables_migration.rs

@@ -2,6 +2,9 @@ use database::db::Db;
 
 #[tokio::main]
 async fn main() {
+    println!("Connecting to the database...");
     let db = Db::connect_to_the_pool().await;
+    println!("Starting migration of tables...");
     db.migrate_tables().await.unwrap();
+    println!("Migration completed.");
 }

database/src/db.rs

@@ -32,10 +32,4 @@ impl Db {
     pub async fn migrate_tables(&self) -> Result<(), sqlx::migrate::MigrateError> {
         migrate!("./migrations").run(&self.connection_pool).await
     }
-
-    pub async fn truncate_table(&self, table_name: &str) -> Result<(), sqlx::Error> {
-        let query = format!("TRUNCATE TABLE {table_name}");
-        sqlx::query(&query).execute(&self.connection_pool).await?;
-        Ok(())
-    }
 }

database/src/lib.rs

@@ -1,2 +1,3 @@
 pub mod db;
+pub mod structs;
 pub mod tables;

database/src/structs/client_data.rs

@@ -0,0 +1,8 @@
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ClientData {
+    pub client_id: Option<String>,
+    pub device: Option<String>,
+    pub metadata: Option<String>,
+    pub notification_endpoint: Option<String>,
+    pub connected_at: u64, // Timestamp of when the client connected to the session
+}

database/src/structs/consts.rs

@@ -0,0 +1,3 @@
+pub const LAST_24_HOURS: &str = "EXTRACT(EPOCH FROM NOW() - INTERVAL '1 day')::BIGINT * 1000";
+pub const LAST_7_DAYS: &str = "EXTRACT(EPOCH FROM NOW() - INTERVAL '7 days')::BIGINT * 1000";
+pub const LAST_30_DAYS: &str = "EXTRACT(EPOCH FROM NOW() - INTERVAL '30 days')::BIGINT * 1000";

database/src/structs/mod.rs

@@ -0,0 +1,4 @@
+pub mod client_data;
+pub mod consts;
+pub mod request_status;
+pub mod subscription;

database/src/structs/request_status.rs

@@ -0,0 +1,10 @@
+use sqlx::Type;
+
+#[derive(Clone, Debug, Eq, PartialEq, Type)]
+#[sqlx(type_name = "request_status_enum")]
+pub enum RequestStatus {
+    Pending,
+    Completed,
+    Rejected,
+    TimedOut,
+}

database/src/structs/subscription.rs

@@ -0,0 +1,9 @@
+use sqlx::Type;
+
+#[derive(Clone, Debug, Eq, PartialEq, Type)]
+#[sqlx(type_name = "subscription")]
+pub struct Subscription {
+    pub subscription_type: String,
+    pub valid_from: i64,
+    pub valid_till: i64,
+}

database/src/tables/mod.rs

@@ -1 +1,5 @@
 pub mod registered_app;
+pub mod requests;
+pub mod sessions;
+pub mod test_utils;
+pub mod utils;

database/src/tables/registered_app/select.rs

@@ -1,5 +1,6 @@
 use super::table_struct::{RegisteredApp, REGISTERED_APPS_TABLE_NAME};
-use crate::db::Db;
+use crate::tables::requests::table_struct::REQUESTS_TABLE_NAME;
+use crate::{db::Db, tables::requests::table_struct::Request};
 use sqlx::query_as;
 
 impl Db {
@@ -15,4 +16,41 @@ impl Db {
             .fetch_one(&self.connection_pool)
             .await;
     }
+
+    pub async fn get_requests_by_app_id(
+        &self,
+        app_id: &String,
+    ) -> Result<Vec<Request>, sqlx::Error> {
+        let query = format!(
+            "SELECT r.* FROM {REQUESTS_TABLE_NAME} r 
+            INNER JOIN sessions s ON r.session_id = s.session_id 
+            WHERE s.app_id = $1
+            ORDER BY r.creation_timestamp DESC"
+        );
+        let typed_query = query_as::<_, Request>(&query);
+
+        return typed_query
+            .bind(&app_id)
+            .fetch_all(&self.connection_pool)
+            .await;
+    }
+
+    pub async fn get_requests_by_app_id_with_filter(
+        &self,
+        app_id: &String,
+        filter: &str,
+    ) -> Result<Vec<Request>, sqlx::Error> {
+        let query = format!(
+            "SELECT r.* FROM {REQUESTS_TABLE_NAME} r 
+            INNER JOIN sessions s ON r.session_id = s.session_id 
+            WHERE s.app_id = $1 AND creation_timestamp >= {filter}
+            ORDER BY r.creation_timestamp DESC"
+        );
+        let typed_query = query_as::<_, Request>(&query);
+
+        return typed_query
+            .bind(&app_id)
+            .fetch_all(&self.connection_pool)
+            .await;
+    }
 }

database/src/tables/registered_app/table_struct.rs

@@ -1,12 +1,5 @@
-use sqlx::{postgres::PgRow, FromRow, Row, Type};
-
-// TODO move later to a common place
-#[derive(Clone, Debug, Eq, PartialEq, Type)]
-#[sqlx(type_name = "subscription")]
-pub struct Subscription {
-    pub email: String,
-    pub subscribed_at: i64,
-}
+use crate::structs::subscription::Subscription;
+use sqlx::{postgres::PgRow, FromRow, Row};
 
 pub const REGISTERED_APPS_TABLE_NAME: &str = "registered_apps";
 pub const REGISTERED_APPS_KEYS: &str = "app_id, app_name, whitelisted_domains, subscription, ack_public_keys, email, registration_timestamp, pass_hash";