feat: add subscriptions to database and the related flows

Makefile

@@ -35,3 +35,5 @@ deploy-k8s:
 	kubectl apply -f ./k8s/migrations.yaml
 	kubectl apply -f ./k8s/deployment.yaml
 
+reset-k8s:
+	kubectl delete all --all

docker-compose.yaml

@@ -110,6 +110,7 @@ services:
       RUST_LOG: debug
       CORS_ORIGIN: http://localhost:3000
       PORT: 8001
+      JWT_SECRET: users-secret-1234
     depends_on:
       postgres:
         condition: service_healthy

frontend/src/components/Feeds.tsx

@@ -1,20 +1,74 @@
-import React, { useEffect, useState } from 'react';
-import api, { Feeds } from '../services/api';
+import React, { useCallback, useEffect, useState } from 'react';
+import api, { Feeds, Subscription } from '../services/api';
 
 const FeedsComponent = () => {
   const [feeds, setFeeds] = useState<Feeds>();
+  const [subscriptions, setSubscriptions] = useState<{
+    [key: string]: boolean;
+  }>({});
 
   useEffect(() => {
     api.feeds().then(setFeeds).catch(console.log);
+    api
+      .subscriptions()
+      .then((subscriptions) => {
+        setSubscriptions(
+          subscriptions.reduce((final, subscription: Subscription) => {
+            final[subscription.feed_id] = true;
+            return final;
+          }, {} as { [key: string]: boolean })
+        );
+      })
+      .catch(console.log);
   }, []);
 
+  const subscribe = useCallback(
+    (feedId: string) => {
+      api.subscribe(feedId).then(() => {
+        setSubscriptions({
+          ...subscriptions,
+          [feedId]: true,
+        });
+      });
+    },
+    [subscriptions]
+  );
+
+  const unsubscribe = useCallback(
+    (feedId: string) => {
+      api.unsubscribe(feedId).then(() => {
+        const newSubscriptions = { ...subscriptions };
+        delete newSubscriptions[feedId];
+        setSubscriptions(newSubscriptions);
+      });
+    },
+    [subscriptions]
+  );
+
   return (
     <>
       {feeds
         ? feeds.map((feed) => {
             return (
               <div>
-                {feed.title} <button>Subscribe</button>
+                {feed.title}{' '}
+                {subscriptions[feed.id] ? (
+                  <button
+                    onClick={() => {
+                      unsubscribe(feed.id);
+                    }}
+                  >
+                    Unsubscribe
+                  </button>
+                ) : (
+                  <button
+                    onClick={() => {
+                      subscribe(feed.id);
+                    }}
+                  >
+                    Subscribe
+                  </button>
+                )}
               </div>
             );
           })

frontend/src/services/api.ts

@@ -15,6 +15,11 @@ export type Feeds = Array<{
   publish_date: string;
 }>;
 
+export type Subscription = {
+  user_id: string;
+  feed_id: string;
+};
+
 class API {
   token: string = '';
 
@@ -87,6 +92,23 @@ class API {
     return this._doRequest('/feeds') as any;
   }
 
+  subscriptions(): Promise<Subscription[]> {
+    return this._doRequest('/subscriptions') as any;
+  }
+
+  subscribe(feedId: string): Promise<Subscription> {
+    return this._doRequest('/subscriptions', {
+      method: 'POST',
+      body: JSON.stringify({ feed_id: feedId }),
+    }) as any;
+  }
+
+  unsubscribe(feedId: string): Promise<number> {
+    return this._doRequest(`/subscriptions?feed_id=${feedId}`, {
+      method: 'DELETE',
+    }) as any;
+  }
+
   connectWs(onMessage: (event: MessageEvent<any>) => void) {
     // const socket = new WebSocket('ws://localhost:8000/ws');
     const socket = new WebSocket(`ws://${window.location.host}/ws`);

frontend/src/setupProxy.js

@@ -12,6 +12,16 @@ module.exports = function (app) {
         },
       })
     )
+    .use(
+      '/api/subscriptions',
+      createProxyMiddleware({
+        target: 'http://news:8001',
+        changeOrigin: true,
+        pathRewrite: {
+          '^/api/': '/',
+        },
+      })
+    )
     .use(
       '/api/feeds',
       createProxyMiddleware({

news/Cargo.toml

@@ -7,14 +7,19 @@ edition = "2021"
 
 [dependencies]
 bytes = "1.5.0"
-diesel = { version="2.1.1", features=["postgres","r2d2","uuid","chrono"]}
+diesel = { version = "2.1.1", features = [
+  "postgres",
+  "r2d2",
+  "uuid",
+  "chrono",
+] }
 feed-rs = "1.3.0"
 futures = "0.3.28"
 reqwest = "0.11.20"
-serde = { version="1.0.188", features=["derive"] }
-tokio = { version="1.32.0", features=["full"] }
+serde = { version = "1.0.188", features = ["derive"] }
+tokio = { version = "1.32.0", features = ["full"] }
 uuid = { version = "1.4.1", features = ["v4", "serde"] }
-utils={ path = "../utils", features = ["broker","database"] }
+utils = { path = "../utils", features = ["broker", "database"] }
 chrono = "0.4.31"
 log = "0.4.20"
 tokio-cron-scheduler = "0.9.4"
@@ -24,3 +29,4 @@ mockall = "0.11.4"
 async-trait = "0.1.73"
 actix-rt = "2.9.0"
 serde_json = "1.0.107"
+validator = { version = "0.16.1", features = ["derive"] }

news/migrations/2023-09-23-202817_create_subscriptions/down.sql

@@ -0,0 +1 @@
+DROP TABLE subscriptions;

news/migrations/2023-09-23-202817_create_subscriptions/up.sql

@@ -0,0 +1,7 @@
+CREATE TABLE subscriptions (
+    feed_id UUID NOT NULL,
+    user_id UUID NOT NULL,
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    FOREIGN KEY (feed_id) REFERENCES feeds (id),
+    PRIMARY KEY (feed_id, user_id)
+);

news/src/config.rs

@@ -1,9 +1,18 @@
+use utils::http::middlewares::jwt_auth::JwtMiddlewareConfig;
+
 #[derive(Debug, Clone)]
 pub struct Config {
     pub cors_origin: String,
     pub database_url: String,
     pub logs_path: String,
     pub server_port: String,
+    pub jwt_secret: String,
+}
+
+impl JwtMiddlewareConfig for Config {
+    fn get_jwt_secret(&self) -> String {
+        return self.jwt_secret.clone();
+    }
 }
 
 impl Config {
@@ -12,12 +21,14 @@ impl Config {
         let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL must be set");
         let logs_path = std::env::var("LOGS_PATH").unwrap_or_else(|_| String::from(""));
         let server_port = std::env::var("PORT").unwrap_or_else(|_| String::from("8000"));
+        let jwt_secret = std::env::var("JWT_SECRET").expect("JWT_SECRET must be set");
 
         Config {
             cors_origin,
             database_url,
             logs_path,
             server_port,
+            jwt_secret,
         }
     }
 }

news/src/handlers/mod.rs

@@ -1,2 +1,3 @@
 pub mod feeds;
 pub mod news;
+pub mod subscriptions;

news/src/handlers/subscriptions.rs

@@ -0,0 +1,104 @@
+use actix_web::{delete, get, web, HttpRequest, HttpResponse};
+use actix_web::{post, HttpMessage};
+use log::error;
+use serde::{Deserialize, Serialize};
+use std::str::FromStr;
+use utils::{error::CommonError, http::middlewares::jwt_auth::JwtMiddleware};
+use uuid::Uuid;
+use validator::Validate;
+
+use crate::models::subscription::Subscription;
+use crate::repositories::subscription_repository::SubscriptionRepository;
+
+#[derive(Debug, Serialize, Deserialize, Validate)]
+pub struct CreateSubscriptionPayload {
+    #[validate(required)]
+    pub feed_id: Option<String>,
+}
+
+#[derive(Deserialize)]
+struct DeleteSubscriptionPayload {
+    feed_id: Option<String>,
+}
+
+#[get("/subscriptions")]
+async fn get_subscriptions(
+    r: HttpRequest,
+    subscription_repo: web::Data<dyn SubscriptionRepository>,
+    _: JwtMiddleware,
+) -> HttpResponse {
+    let user_id = *r.extensions().get::<uuid::Uuid>().unwrap();
+
+    let result = subscription_repo.list_by_user(user_id);
+
+    match result {
+        Err(err) => {
+            error!("failed getting subscriptions: {}", CommonError::from(err));
+            HttpResponse::InternalServerError().finish()
+        }
+        Ok(feeds) => HttpResponse::Ok().json(feeds),
+    }
+}
+
+#[post("/subscriptions")]
+async fn create_subscription(
+    r: HttpRequest,
+    subscription_repo: web::Data<dyn SubscriptionRepository>,
+    payload: Option<web::Json<CreateSubscriptionPayload>>,
+    _: JwtMiddleware,
+) -> HttpResponse {
+    let user_id = *r.extensions().get::<uuid::Uuid>().unwrap();
+
+    if payload.is_none() {
+        return HttpResponse::BadRequest().body("empty body");
+    }
+
+    let payload = payload.unwrap();
+    if let Err(err) = payload.validate() {
+        return HttpResponse::BadRequest().json(err);
+    }
+
+    let CreateSubscriptionPayload { feed_id } = payload.into_inner();
+
+    if feed_id.is_none() {
+        return HttpResponse::BadRequest().body("Missing 'feed_id' in body");
+    }
+
+    let feed_id = Uuid::from_str(&feed_id.unwrap().to_string()).unwrap();
+
+    let result = subscription_repo.create(&Subscription { feed_id, user_id });
+
+    match result {
+        Err(err) => {
+            error!("failed creating subscription: {}", CommonError::from(err));
+            HttpResponse::InternalServerError().finish()
+        }
+        Ok(subscription) => HttpResponse::Ok().json(subscription),
+    }
+}
+
+#[delete("/subscriptions")]
+async fn delete_subscription(
+    r: HttpRequest,
+    subscription_repo: web::Data<dyn SubscriptionRepository>,
+    _: JwtMiddleware,
+    payload: web::Query<DeleteSubscriptionPayload>,
+) -> HttpResponse {
+    let user_id = *r.extensions().get::<uuid::Uuid>().unwrap();
+
+    if let Some(feed_id) = &payload.feed_id {
+        let feed_id = Uuid::from_str(&feed_id.clone()).unwrap();
+
+        let result = subscription_repo.delete(feed_id, user_id);
+
+        match result {
+            Err(err) => {
+                error!("failed deleting subscription: {}", CommonError::from(err));
+                HttpResponse::InternalServerError().finish()
+            }
+            Ok(subscription) => HttpResponse::Ok().json(subscription),
+        }
+    } else {
+        HttpResponse::BadRequest().body("Missing 'feed_id' query parameter")
+    }
+}