forked from
tobinio.dev/task-line
A simple to-do app focused on tasks that can be completed within a specific time span.
category sse
+197
-76
+84
-75
api/src/routes/categories.rs
+84
-75
api/src/routes/categories.rs
···
1
+
use std::convert::Infallible;
2
+
use std::time::Duration;
3
+
1
4
use axum::extract::Path;
5
+
use axum::response::Sse;
6
+
use axum::response::sse::Event;
2
7
use axum::routing::{delete, post};
3
8
use axum::{Json, Router, extract::State, routing::get};
9
+
use futures::Stream;
4
10
use http::StatusCode;
5
-
use sea_orm::ActiveValue::Set;
6
-
use sea_orm::{ColumnTrait, EntityTrait, ModelTrait, QueryFilter};
11
+
use tokio_stream::StreamExt;
12
+
use tokio_stream::wrappers::BroadcastStream;
7
13
use tracing::warn;
8
-
use types::{Category as CategoryModel, HexColor};
14
+
use types::Category as CategoryModel;
9
15
use uuid::Uuid;
10
16
17
+
use crate::services::categories::CategoryService;
18
+
use crate::services::events;
11
19
use crate::{AppState, auth::User};
12
20
13
-
use crate::entities::{category, prelude::*};
14
-
15
21
pub fn routes(state: AppState) -> Router {
16
22
Router::new()
17
23
.route("/", get(get_all_categories))
18
24
.route("/", post(add_category))
19
25
.route("/{category_uuid}", delete(delete_category))
26
+
.route("/sse", get(sse_handler))
20
27
.with_state(state)
21
28
}
22
29
···
25
32
user: User,
26
33
Json(category): Json<CategoryModel>,
27
34
) -> Result<Json<CategoryModel>, (StatusCode, &'static str)> {
28
-
let existing_category = Category::find_by_id(category.uuid)
29
-
.one(&state.db_connection)
35
+
let existing_category = CategoryService::get_by_id(&user, &state.db_connection, category.uuid)
30
36
.await
31
37
.map_err(|e| {
32
38
warn!("failed to find category: {}", e);
33
39
(StatusCode::INTERNAL_SERVER_ERROR, "failed to find category")
34
40
})?;
35
41
36
-
let category = category::ActiveModel {
37
-
id: Set(category.uuid),
38
-
owner_id: Set(user.uuid),
39
-
name: Set(category.name),
40
-
color: Set(category.color.as_str().to_string()),
41
-
icon: Set(category.icon),
42
-
};
43
-
44
42
let new_category = match existing_category {
45
43
Some(existing_category) => {
46
-
Category::update(category)
47
-
.exec(&state.db_connection)
48
-
.await
49
-
.map_err(|e| {
50
-
warn!("failed to add category: {}", e);
51
-
(StatusCode::INTERNAL_SERVER_ERROR, "failed to add category")
52
-
})?;
53
-
54
-
CategoryModel {
55
-
uuid: existing_category.id,
56
-
name: existing_category.name,
57
-
color: HexColor::from_text(&existing_category.color).unwrap(),
58
-
icon: existing_category.icon,
59
-
}
60
-
}
61
-
None => {
62
-
let new_category = Category::insert(category)
63
-
.exec_with_returning(&state.db_connection)
44
+
CategoryService::update(&user, &state.db_connection, category)
64
45
.await
65
46
.map_err(|e| {
66
47
warn!("failed to add category: {}", e);
67
48
(StatusCode::INTERNAL_SERVER_ERROR, "failed to add category")
68
49
})?;
69
50
70
-
CategoryModel {
71
-
uuid: new_category.id,
72
-
name: new_category.name,
73
-
color: HexColor::from_text(&new_category.color).unwrap(),
74
-
icon: new_category.icon,
75
-
}
51
+
existing_category
76
52
}
53
+
None => CategoryService::create(&user, &state.db_connection, category)
54
+
.await
55
+
.map_err(|e| {
56
+
warn!("failed to add category: {}", e);
57
+
(StatusCode::INTERNAL_SERVER_ERROR, "failed to add category")
58
+
})?,
77
59
};
78
60
61
+
state
62
+
.event_service
63
+
.broadcast(events::Event::Category(user.uuid));
79
64
Ok(Json(new_category))
80
65
}
81
66
···
83
68
state: State<AppState>,
84
69
user: User,
85
70
) -> Result<Json<Vec<CategoryModel>>, (StatusCode, &'static str)> {
86
-
let categories = Category::find()
87
-
.filter(category::Column::OwnerId.eq(user.uuid))
88
-
.all(&state.db_connection)
71
+
let categories = CategoryService::get_all(&user, &state.db_connection)
89
72
.await
90
73
.map_err(|e| {
91
74
warn!("failed to fetch category: {}", e);
···
95
78
)
96
79
})?;
97
80
98
-
Ok(Json(
99
-
categories
100
-
.into_iter()
101
-
.map(|category| CategoryModel {
102
-
uuid: category.id,
103
-
name: category.name,
104
-
color: HexColor::from_text(&category.color).unwrap(),
105
-
icon: category.icon,
106
-
})
107
-
.collect(),
108
-
))
81
+
Ok(Json(categories))
109
82
}
110
83
111
84
async fn delete_category(
···
113
86
Path(category_uuid): Path<Uuid>,
114
87
user: User,
115
88
) -> Result<Json<CategoryModel>, (StatusCode, &'static str)> {
116
-
let deleted_category = Category::find_by_id(category_uuid)
117
-
.filter(category::Column::OwnerId.eq(user.uuid))
118
-
.one(&state.db_connection)
89
+
let to_be_deleted_category =
90
+
CategoryService::get_by_id(&user, &state.db_connection, category_uuid)
91
+
.await
92
+
.map_err(|e| {
93
+
warn!("failed to fetch category: {}", e);
94
+
(
95
+
StatusCode::INTERNAL_SERVER_ERROR,
96
+
"failed to fetch categories",
97
+
)
98
+
})?;
99
+
100
+
let Some(deleted_category) = to_be_deleted_category else {
101
+
return Err((StatusCode::NOT_FOUND, "category not found"));
102
+
};
103
+
104
+
CategoryService::delete_by_id(&user, &state.db_connection, category_uuid)
119
105
.await
120
-
.map_err(|e| {
121
-
warn!("failed to fetch category: {}", e);
106
+
.map_err(|_| {
122
107
(
123
108
StatusCode::INTERNAL_SERVER_ERROR,
124
-
"failed to fetch categories",
109
+
"failed to delete category",
125
110
)
126
111
})?;
127
112
128
-
let Some(category) = deleted_category else {
129
-
return Err((StatusCode::NOT_FOUND, "category not found"));
130
-
};
113
+
state
114
+
.event_service
115
+
.broadcast(events::Event::Category(user.uuid));
116
+
Ok(Json(deleted_category))
117
+
}
118
+
119
+
async fn sse_handler(
120
+
state: State<AppState>,
121
+
user: User,
122
+
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
123
+
let stream = BroadcastStream::new(state.event_service.subscribe())
124
+
.filter(move |event| {
125
+
let Ok(event) = event else {
126
+
return false;
127
+
};
131
128
132
-
let deleted_category = CategoryModel {
133
-
uuid: category.id,
134
-
name: category.name.to_string(),
135
-
color: HexColor::from_text(&category.color).unwrap(),
136
-
icon: category.icon.to_string(),
137
-
};
129
+
match event {
130
+
events::Event::Tag(_) => false,
131
+
events::Event::Category(user_uuid) => user_uuid == &user.uuid,
132
+
}
133
+
})
134
+
.then(move |_| {
135
+
let db = state.db_connection.clone();
136
+
let user = user.clone();
138
137
139
-
category.delete(&state.db_connection).await.map_err(|_| {
140
-
(
141
-
StatusCode::INTERNAL_SERVER_ERROR,
142
-
"failed to delete category",
143
-
)
144
-
})?;
138
+
async move {
139
+
CategoryService::get_all(&user, &db)
140
+
.await
141
+
.map_err(|e| {
142
+
warn!("failed to fetch tag: {}", e);
143
+
"failed to fetch tags"
144
+
})
145
+
.map(|tags| Event::default().data(serde_json::to_string(&tags).unwrap()))
146
+
.unwrap()
147
+
}
148
+
})
149
+
.map(Ok);
145
150
146
-
Ok(Json(deleted_category))
151
+
Sse::new(stream).keep_alive(
152
+
axum::response::sse::KeepAlive::new()
153
+
.interval(Duration::from_secs(1))
154
+
.text("keep-alive-text"),
155
+
)
147
156
}
+110
api/src/services/categories.rs
+110
api/src/services/categories.rs
···
1
+
use crate::{
2
+
auth::User,
3
+
entities::{category, prelude::*},
4
+
};
5
+
6
+
use sea_orm::{
7
+
ActiveValue::Set, ColumnTrait, DatabaseConnection, DeleteResult, EntityTrait, QueryFilter,
8
+
};
9
+
use types::{Category as CategoryModel, HexColor};
10
+
use uuid::Uuid;
11
+
12
+
pub struct CategoryService;
13
+
14
+
impl CategoryService {
15
+
pub async fn get_all(
16
+
user: &User,
17
+
db: &DatabaseConnection,
18
+
) -> Result<Vec<CategoryModel>, sea_orm::DbErr> {
19
+
Category::find()
20
+
.filter(category::Column::OwnerId.eq(user.uuid))
21
+
.all(db)
22
+
.await
23
+
.map(|categoryies| {
24
+
categoryies
25
+
.into_iter()
26
+
.map(|category| category.into())
27
+
.collect()
28
+
})
29
+
}
30
+
31
+
pub async fn get_by_id(
32
+
user: &User,
33
+
db: &DatabaseConnection,
34
+
category_id: Uuid,
35
+
) -> Result<Option<CategoryModel>, sea_orm::DbErr> {
36
+
Category::find_by_id(category_id)
37
+
.filter(category::Column::OwnerId.eq(user.uuid))
38
+
.one(db)
39
+
.await
40
+
.map(|category| category.map(|category| category.into()))
41
+
}
42
+
43
+
pub async fn delete_by_id(
44
+
user: &User,
45
+
db: &DatabaseConnection,
46
+
category_id: Uuid,
47
+
) -> Result<DeleteResult, sea_orm::DbErr> {
48
+
Category::delete_by_id(category_id)
49
+
.filter(category::Column::OwnerId.eq(user.uuid))
50
+
.exec(db)
51
+
.await
52
+
}
53
+
54
+
pub async fn create(
55
+
user: &User,
56
+
db: &DatabaseConnection,
57
+
category: CategoryModel,
58
+
) -> Result<CategoryModel, sea_orm::DbErr> {
59
+
Category::insert(category::ActiveModel {
60
+
id: Set(category.uuid),
61
+
owner_id: Set(user.uuid),
62
+
name: Set(category.name),
63
+
color: Set(category.color.as_str().to_string()),
64
+
icon: Set(category.icon),
65
+
})
66
+
.exec_with_returning(db)
67
+
.await
68
+
.map(|category| category.into())
69
+
}
70
+
71
+
pub async fn update(
72
+
user: &User,
73
+
db: &DatabaseConnection,
74
+
category: CategoryModel,
75
+
) -> Result<CategoryModel, sea_orm::DbErr> {
76
+
Category::update(category::ActiveModel {
77
+
id: Set(category.uuid),
78
+
owner_id: Set(user.uuid),
79
+
name: Set(category.name),
80
+
color: Set(category.color.as_str().to_string()),
81
+
icon: Set(category.icon),
82
+
})
83
+
.filter(category::Column::OwnerId.eq(user.uuid))
84
+
.exec(db)
85
+
.await
86
+
.map(|category| category.into())
87
+
}
88
+
}
89
+
90
+
impl From<category::Model> for CategoryModel {
91
+
fn from(value: category::Model) -> Self {
92
+
CategoryModel {
93
+
uuid: value.id,
94
+
name: value.name,
95
+
color: HexColor::from_text(&value.color).unwrap(),
96
+
icon: value.icon,
97
+
}
98
+
}
99
+
}
100
+
101
+
impl From<&category::Model> for CategoryModel {
102
+
fn from(value: &category::Model) -> Self {
103
+
CategoryModel {
104
+
uuid: value.id,
105
+
name: value.name.to_string(),
106
+
color: HexColor::from_text(&value.color).unwrap(),
107
+
icon: value.icon.to_string(),
108
+
}
109
+
}
110
+
}
+1
api/src/services/events.rs
+1
api/src/services/events.rs