hexagonal refactor
This commit is contained in:
+63
-853
@@ -1,224 +1,32 @@
|
||||
mod db;
|
||||
mod domain;
|
||||
mod http;
|
||||
mod hub;
|
||||
mod ports;
|
||||
mod security;
|
||||
mod seed;
|
||||
mod services;
|
||||
mod sqlite;
|
||||
mod views;
|
||||
|
||||
use std::time::Duration;
|
||||
use std::{env, future::Future, path::Path as FilePath, sync::Arc};
|
||||
use std::env;
|
||||
use std::path::Path as FilePath;
|
||||
use std::sync::Arc;
|
||||
|
||||
use argon2::{
|
||||
Argon2,
|
||||
password_hash::{PasswordHash, PasswordHasher, PasswordVerifier, SaltString, rand_core::OsRng},
|
||||
};
|
||||
use axum::{
|
||||
Router,
|
||||
extract::{
|
||||
Form, FromRequest, FromRequestParts, Path, Query, Request, State,
|
||||
ws::{Message, WebSocket, WebSocketUpgrade},
|
||||
},
|
||||
http::{HeaderMap, HeaderValue, StatusCode, header, request::Parts},
|
||||
middleware::{self, Next},
|
||||
response::{Html, IntoResponse, Redirect, Response},
|
||||
routing::{get, post},
|
||||
};
|
||||
use futures_util::{SinkExt, StreamExt};
|
||||
use serde::{Deserialize, de::DeserializeOwned};
|
||||
use thiserror::Error;
|
||||
use tokio::net::TcpListener;
|
||||
use tower_http::{services::ServeDir, trace::TraceLayer};
|
||||
use tracing::{error, info, warn};
|
||||
use tracing::{info, warn};
|
||||
|
||||
use crate::{
|
||||
db::{Database, DbError, SessionUser},
|
||||
hub::{Hub, HubEvent},
|
||||
use crate::http::{AppState, build_router};
|
||||
use crate::hub::InMemoryHub;
|
||||
use crate::ports::{
|
||||
CategoryRepository, InvitationRepository, ItemRepository, ListRepository, PasswordHasher,
|
||||
RealtimeNotifier, SessionRepository, TokenGenerator, UserRepository,
|
||||
};
|
||||
use crate::security::{Argon2PasswordHasher, RandomTokenGenerator};
|
||||
use crate::services::{AuthService, InvitationService, ListService, RegistrationMode};
|
||||
use crate::sqlite::{
|
||||
SqliteCategoryRepository, SqliteInvitationRepository, SqliteItemRepository,
|
||||
SqliteListRepository, SqliteSessionRepository, SqliteDatabase, SqliteUserRepository,
|
||||
};
|
||||
|
||||
#[derive(Clone)]
|
||||
struct AppState {
|
||||
db: Database,
|
||||
hub: Arc<Hub>,
|
||||
cookie_secure: bool,
|
||||
public_base_url: String,
|
||||
registration_mode: RegistrationMode,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, PartialEq, Eq)]
|
||||
enum RegistrationMode {
|
||||
Open,
|
||||
InviteOnly,
|
||||
}
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
enum AppError {
|
||||
#[error("database error")]
|
||||
Database(#[from] DbError),
|
||||
#[error("bad request: {0}")]
|
||||
BadRequest(String),
|
||||
#[error("not found")]
|
||||
NotFound,
|
||||
#[error("internal error: {0}")]
|
||||
Internal(String),
|
||||
}
|
||||
|
||||
impl IntoResponse for AppError {
|
||||
fn into_response(self) -> Response {
|
||||
let (status, heading, message): (StatusCode, &str, String) = match self {
|
||||
Self::Database(DbError::NotFound) | Self::NotFound => (
|
||||
StatusCode::NOT_FOUND,
|
||||
"Not found",
|
||||
"We could not find that page or list.".into(),
|
||||
),
|
||||
Self::Database(DbError::Conflict) => (
|
||||
StatusCode::CONFLICT,
|
||||
"Already exists",
|
||||
"That value is already in use.".into(),
|
||||
),
|
||||
Self::BadRequest(message) => {
|
||||
warn!(reason = %message, "request rejected");
|
||||
(StatusCode::BAD_REQUEST, "Check that again", message)
|
||||
}
|
||||
Self::Database(error) => {
|
||||
error!(%error, "database request failed");
|
||||
(
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
"Something went wrong",
|
||||
"The request could not be completed.".into(),
|
||||
)
|
||||
}
|
||||
Self::Internal(error) => {
|
||||
error!(%error, "request failed");
|
||||
(
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
"Something went wrong",
|
||||
"The request could not be completed.".into(),
|
||||
)
|
||||
}
|
||||
};
|
||||
|
||||
(
|
||||
status,
|
||||
Html(views::error_page(heading, &message).into_string()),
|
||||
)
|
||||
.into_response()
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
struct CurrentUser {
|
||||
session_token: String,
|
||||
session: SessionUser,
|
||||
}
|
||||
|
||||
impl FromRequestParts<AppState> for CurrentUser {
|
||||
type Rejection = Response;
|
||||
|
||||
fn from_request_parts(
|
||||
parts: &mut Parts,
|
||||
state: &AppState,
|
||||
) -> impl Future<Output = Result<Self, Self::Rejection>> + Send {
|
||||
let session_token = cookie_value(&parts.headers, "session");
|
||||
async move {
|
||||
let Some(session_token) = session_token else {
|
||||
return Err(Redirect::to("/login").into_response());
|
||||
};
|
||||
|
||||
match state.db.session_user(session_token.clone()).await {
|
||||
Ok(Some(session)) => Ok(Self {
|
||||
session_token,
|
||||
session,
|
||||
}),
|
||||
Ok(None) => Err(Redirect::to("/login").into_response()),
|
||||
Err(error) => Err(AppError::Database(error).into_response()),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
struct LoggedForm<T>(T);
|
||||
|
||||
impl<S, T> FromRequest<S> for LoggedForm<T>
|
||||
where
|
||||
S: Send + Sync,
|
||||
T: DeserializeOwned + Send,
|
||||
{
|
||||
type Rejection = Response;
|
||||
|
||||
fn from_request(
|
||||
request: Request,
|
||||
state: &S,
|
||||
) -> impl Future<Output = Result<Self, Self::Rejection>> + Send {
|
||||
let method = request.method().clone();
|
||||
let uri = request.uri().clone();
|
||||
async move {
|
||||
match Form::<T>::from_request(request, state).await {
|
||||
Ok(Form(value)) => Ok(Self(value)),
|
||||
Err(rejection) => {
|
||||
warn!(
|
||||
%method,
|
||||
%uri,
|
||||
rejection = ?rejection,
|
||||
"request form deserialization failed"
|
||||
);
|
||||
Err(rejection.into_response())
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct InviteQuery {
|
||||
invite: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct RegisterForm {
|
||||
display_name: String,
|
||||
email: String,
|
||||
password: String,
|
||||
invite: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct LoginForm {
|
||||
email: String,
|
||||
password: String,
|
||||
invite: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct CreateListForm {
|
||||
name: String,
|
||||
csrf: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct ItemForm {
|
||||
name: String,
|
||||
quantity: String,
|
||||
#[serde(default)]
|
||||
note: String,
|
||||
#[serde(default)]
|
||||
category_id: Option<String>,
|
||||
csrf: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct CheckForm {
|
||||
checked: String,
|
||||
csrf: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct CsrfForm {
|
||||
csrf: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct CategoryForm {
|
||||
name: String,
|
||||
csrf: String,
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
@@ -248,37 +56,52 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
}
|
||||
};
|
||||
|
||||
let state = AppState {
|
||||
db: Database::open(database_path)?,
|
||||
hub: Arc::new(Hub::default()),
|
||||
cookie_secure,
|
||||
public_base_url,
|
||||
// Build the adapters (ports) and wire them into application services.
|
||||
let db = SqliteDatabase::open(&database_path).await?;
|
||||
let users: Arc<dyn UserRepository> = Arc::new(SqliteUserRepository);
|
||||
let sessions: Arc<dyn SessionRepository> = Arc::new(SqliteSessionRepository);
|
||||
let lists: Arc<dyn ListRepository> = Arc::new(SqliteListRepository);
|
||||
let categories: Arc<dyn CategoryRepository> = Arc::new(SqliteCategoryRepository);
|
||||
let items: Arc<dyn ItemRepository> = Arc::new(SqliteItemRepository);
|
||||
let invitations: Arc<dyn InvitationRepository> = Arc::new(SqliteInvitationRepository);
|
||||
let hasher: Arc<dyn PasswordHasher> = Arc::new(Argon2PasswordHasher);
|
||||
let tokens: Arc<dyn TokenGenerator> = Arc::new(RandomTokenGenerator);
|
||||
let realtime: Arc<dyn RealtimeNotifier> = Arc::new(InMemoryHub::default());
|
||||
|
||||
let auth = Arc::new(AuthService::new(
|
||||
db.clone(),
|
||||
Arc::clone(&users),
|
||||
Arc::clone(&sessions),
|
||||
Arc::clone(&invitations),
|
||||
Arc::clone(&hasher),
|
||||
registration_mode,
|
||||
};
|
||||
));
|
||||
let lists_service = Arc::new(ListService::new(
|
||||
db.clone(),
|
||||
Arc::clone(&lists),
|
||||
Arc::clone(&categories),
|
||||
Arc::clone(&items),
|
||||
Arc::clone(&realtime),
|
||||
));
|
||||
let invitations_service = Arc::new(InvitationService::new(
|
||||
db.clone(),
|
||||
Arc::clone(&invitations),
|
||||
Arc::clone(&tokens),
|
||||
));
|
||||
|
||||
let seed_path = env::var("SEED_CONFIG").unwrap_or_else(|_| "seed.json".into());
|
||||
seed::seed_if_needed(&state.db, FilePath::new(&seed_path)).await;
|
||||
seed::seed_if_needed(&db, &users, &hasher, FilePath::new(&seed_path)).await;
|
||||
|
||||
let app = Router::new()
|
||||
.route("/", get(home))
|
||||
.route("/login", get(login_page).post(login))
|
||||
.route("/register", get(register_page).post(register))
|
||||
.route("/logout", post(logout))
|
||||
.route("/lists", get(lists_page).post(create_list))
|
||||
.route("/lists/{list_id}", get(list_page))
|
||||
.route("/lists/{list_id}/items", post(add_item))
|
||||
.route("/lists/{list_id}/items/{item_id}/check", post(check_item))
|
||||
.route("/lists/{list_id}/items/{item_id}/edit", post(edit_item))
|
||||
.route("/lists/{list_id}/items/{item_id}/delete", post(delete_item))
|
||||
.route("/lists/{list_id}/categories", post(create_category))
|
||||
.route("/invitations", post(create_invitation))
|
||||
.route("/lists/{list_id}/stream", get(list_stream))
|
||||
.route("/invite/{token}", get(invitation_page))
|
||||
.route("/invite/{token}/accept", post(accept_invitation))
|
||||
.nest_service("/static", ServeDir::new("static"))
|
||||
.layer(TraceLayer::new_for_http())
|
||||
.layer(middleware::from_fn(log_response_status))
|
||||
.with_state(state);
|
||||
let state = AppState {
|
||||
auth,
|
||||
lists: lists_service,
|
||||
invitations: invitations_service,
|
||||
realtime,
|
||||
cookie_secure,
|
||||
public_base_url,
|
||||
};
|
||||
|
||||
let app = build_router(state);
|
||||
|
||||
let listener = TcpListener::bind(&bind_address).await?;
|
||||
info!(address = %bind_address, "sustenance listening");
|
||||
@@ -314,616 +137,3 @@ async fn shutdown_signal() {
|
||||
|
||||
info!("signal received; starting graceful shutdown");
|
||||
}
|
||||
|
||||
async fn home() -> Redirect {
|
||||
Redirect::to("/lists")
|
||||
}
|
||||
|
||||
async fn log_response_status(request: Request, next: Next) -> Response {
|
||||
let method = request.method().clone();
|
||||
let uri = request.uri().clone();
|
||||
let response = next.run(request).await;
|
||||
let status = response.status();
|
||||
|
||||
if status.is_server_error() {
|
||||
error!(%method, %uri, %status, "request returned server error");
|
||||
} else if status.is_client_error() {
|
||||
warn!(%method, %uri, %status, "request returned client error");
|
||||
}
|
||||
|
||||
response
|
||||
}
|
||||
|
||||
async fn login_page(Query(query): Query<InviteQuery>) -> Result<Response, AppError> {
|
||||
Ok(html_response(views::login_page(
|
||||
None,
|
||||
query.invite.as_deref(),
|
||||
)))
|
||||
}
|
||||
|
||||
async fn register_page(
|
||||
State(state): State<AppState>,
|
||||
Query(query): Query<InviteQuery>,
|
||||
) -> Result<Response, AppError> {
|
||||
if can_register(&state, query.invite.as_deref()).await? {
|
||||
Ok(html_response(views::register_page(
|
||||
None,
|
||||
query.invite.as_deref(),
|
||||
)))
|
||||
} else {
|
||||
Ok(status_html_response(
|
||||
StatusCode::FORBIDDEN,
|
||||
views::registration_closed_page(),
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
async fn register(
|
||||
State(state): State<AppState>,
|
||||
LoggedForm(form): LoggedForm<RegisterForm>,
|
||||
) -> Result<Response, AppError> {
|
||||
if !can_register(&state, form.invite.as_deref()).await? {
|
||||
return Ok(status_html_response(
|
||||
StatusCode::FORBIDDEN,
|
||||
views::registration_closed_page(),
|
||||
));
|
||||
}
|
||||
let display_name = form.display_name.trim().to_owned();
|
||||
let email = form.email.trim().to_lowercase();
|
||||
if display_name.is_empty() || display_name.chars().count() > 50 {
|
||||
return Ok(html_response(views::register_page(
|
||||
Some("Enter a name between 1 and 50 characters."),
|
||||
form.invite.as_deref(),
|
||||
)));
|
||||
}
|
||||
if !email.contains('@') || email.len() > 200 {
|
||||
return Ok(html_response(views::register_page(
|
||||
Some("Enter a valid email address."),
|
||||
form.invite.as_deref(),
|
||||
)));
|
||||
}
|
||||
|
||||
let password = form.password;
|
||||
let password_hash = tokio::task::spawn_blocking(move || hash_password(&password))
|
||||
.await
|
||||
.map_err(|error| AppError::Internal(error.to_string()))?
|
||||
.map_err(AppError::Internal)?;
|
||||
let user = match state
|
||||
.db
|
||||
.create_user(email, display_name, password_hash)
|
||||
.await
|
||||
{
|
||||
Ok(user) => user,
|
||||
Err(DbError::Conflict) => {
|
||||
return Ok(html_response(views::register_page(
|
||||
Some("An account with that email already exists."),
|
||||
form.invite.as_deref(),
|
||||
)));
|
||||
}
|
||||
Err(error) => return Err(AppError::Database(error)),
|
||||
};
|
||||
|
||||
let (session_token, _) = state.db.create_session(user.id).await?;
|
||||
let destination = form
|
||||
.invite
|
||||
.filter(|invite| !invite.is_empty())
|
||||
.map(|invite| format!("/invite/{invite}"))
|
||||
.unwrap_or_else(|| "/lists".into());
|
||||
let mut response = Redirect::to(&destination).into_response();
|
||||
set_session_cookie(&mut response, &session_token, state.cookie_secure);
|
||||
Ok(response)
|
||||
}
|
||||
|
||||
async fn login(
|
||||
State(state): State<AppState>,
|
||||
LoggedForm(form): LoggedForm<LoginForm>,
|
||||
) -> Result<Response, AppError> {
|
||||
let email = form.email.trim().to_lowercase();
|
||||
let Some((user, password_hash)) = state.db.find_user_by_email(email).await? else {
|
||||
return Ok(html_response(views::login_page(
|
||||
Some("Email or password is incorrect."),
|
||||
form.invite.as_deref(),
|
||||
)));
|
||||
};
|
||||
|
||||
let password = form.password;
|
||||
let valid = tokio::task::spawn_blocking(move || verify_password(&password, &password_hash))
|
||||
.await
|
||||
.map_err(|error| AppError::Internal(error.to_string()))?
|
||||
.map_err(AppError::Internal)?;
|
||||
if !valid {
|
||||
return Ok(html_response(views::login_page(
|
||||
Some("Email or password is incorrect."),
|
||||
form.invite.as_deref(),
|
||||
)));
|
||||
}
|
||||
|
||||
let (session_token, _) = state.db.create_session(user.id).await?;
|
||||
let destination = form
|
||||
.invite
|
||||
.filter(|invite| !invite.is_empty())
|
||||
.map(|invite| format!("/invite/{invite}"))
|
||||
.unwrap_or_else(|| "/lists".into());
|
||||
let mut response = Redirect::to(&destination).into_response();
|
||||
set_session_cookie(&mut response, &session_token, state.cookie_secure);
|
||||
Ok(response)
|
||||
}
|
||||
|
||||
async fn logout(State(state): State<AppState>, user: CurrentUser) -> Result<Response, AppError> {
|
||||
state.db.delete_session(user.session_token).await?;
|
||||
let mut response = Redirect::to("/login").into_response();
|
||||
clear_session_cookie(&mut response, state.cookie_secure);
|
||||
Ok(response)
|
||||
}
|
||||
|
||||
async fn lists_page(
|
||||
State(state): State<AppState>,
|
||||
user: CurrentUser,
|
||||
) -> Result<Response, AppError> {
|
||||
let lists = state.db.list_summaries().await?;
|
||||
Ok(html_response(views::lists_page(
|
||||
&user.session.user,
|
||||
&lists,
|
||||
&user.session.csrf_token,
|
||||
)))
|
||||
}
|
||||
|
||||
async fn create_list(
|
||||
State(state): State<AppState>,
|
||||
user: CurrentUser,
|
||||
LoggedForm(form): LoggedForm<CreateListForm>,
|
||||
) -> Result<Response, AppError> {
|
||||
verify_csrf(&user, &form.csrf)?;
|
||||
let name = form.name.trim().to_owned();
|
||||
if name.is_empty() || name.chars().count() > 80 {
|
||||
return Err(AppError::BadRequest(
|
||||
"List names must be between 1 and 80 characters.".into(),
|
||||
));
|
||||
}
|
||||
let list = state.db.create_list(name).await?;
|
||||
Ok(Redirect::to(&format!("/lists/{}", list.id)).into_response())
|
||||
}
|
||||
|
||||
async fn list_page(
|
||||
State(state): State<AppState>,
|
||||
user: CurrentUser,
|
||||
Path(list_id): Path<i64>,
|
||||
) -> Result<Response, AppError> {
|
||||
let access = require_access(&state, list_id).await?;
|
||||
let items = state.db.items(list_id).await?;
|
||||
let categories = state.db.categories(list_id).await?;
|
||||
let presence = state.hub.presence(list_id).await;
|
||||
Ok(html_response(views::list_page(
|
||||
&user.session.user,
|
||||
&access,
|
||||
&items,
|
||||
&categories,
|
||||
&presence,
|
||||
&user.session.csrf_token,
|
||||
)))
|
||||
}
|
||||
|
||||
async fn add_item(
|
||||
State(state): State<AppState>,
|
||||
user: CurrentUser,
|
||||
Path(list_id): Path<i64>,
|
||||
LoggedForm(form): LoggedForm<ItemForm>,
|
||||
) -> Result<Response, AppError> {
|
||||
verify_csrf(&user, &form.csrf)?;
|
||||
require_access(&state, list_id).await?;
|
||||
let name = form.name.trim().to_owned();
|
||||
let quantity = form.quantity.trim().to_owned();
|
||||
let note = form.note.trim().to_owned();
|
||||
let category_id = parse_category_id(form.category_id);
|
||||
if name.is_empty() || name.chars().count() > 120 {
|
||||
return Err(AppError::BadRequest(
|
||||
"Item names must be between 1 and 120 characters.".into(),
|
||||
));
|
||||
}
|
||||
let revision = state
|
||||
.db
|
||||
.add_item(list_id, name, quantity, note, category_id)
|
||||
.await?;
|
||||
state.hub.publish_list_changed(list_id, revision).await;
|
||||
list_fragment_response(&state, &user, list_id).await
|
||||
}
|
||||
|
||||
async fn check_item(
|
||||
State(state): State<AppState>,
|
||||
user: CurrentUser,
|
||||
Path((list_id, item_id)): Path<(i64, i64)>,
|
||||
LoggedForm(form): LoggedForm<CheckForm>,
|
||||
) -> Result<Response, AppError> {
|
||||
verify_csrf(&user, &form.csrf)?;
|
||||
require_access(&state, list_id).await?;
|
||||
let checked = match form.checked.as_str() {
|
||||
"1" | "true" => true,
|
||||
"0" | "false" => false,
|
||||
_ => return Err(AppError::BadRequest("Invalid checked value.".into())),
|
||||
};
|
||||
let revision = state.db.set_item_checked(list_id, item_id, checked).await?;
|
||||
state.hub.publish_list_changed(list_id, revision).await;
|
||||
list_fragment_response(&state, &user, list_id).await
|
||||
}
|
||||
|
||||
async fn edit_item(
|
||||
State(state): State<AppState>,
|
||||
user: CurrentUser,
|
||||
Path((list_id, item_id)): Path<(i64, i64)>,
|
||||
LoggedForm(form): LoggedForm<ItemForm>,
|
||||
) -> Result<Response, AppError> {
|
||||
verify_csrf(&user, &form.csrf)?;
|
||||
require_access(&state, list_id).await?;
|
||||
let name = form.name.trim().to_owned();
|
||||
if name.is_empty() || name.chars().count() > 120 {
|
||||
return Err(AppError::BadRequest(
|
||||
"Item names must be between 1 and 120 characters.".into(),
|
||||
));
|
||||
}
|
||||
let revision = state
|
||||
.db
|
||||
.update_item(
|
||||
list_id,
|
||||
item_id,
|
||||
name,
|
||||
form.quantity.trim().to_owned(),
|
||||
form.note.trim().to_owned(),
|
||||
parse_category_id(form.category_id),
|
||||
)
|
||||
.await?;
|
||||
state.hub.publish_list_changed(list_id, revision).await;
|
||||
list_fragment_response(&state, &user, list_id).await
|
||||
}
|
||||
|
||||
async fn delete_item(
|
||||
State(state): State<AppState>,
|
||||
user: CurrentUser,
|
||||
Path((list_id, item_id)): Path<(i64, i64)>,
|
||||
LoggedForm(form): LoggedForm<CsrfForm>,
|
||||
) -> Result<Response, AppError> {
|
||||
verify_csrf(&user, &form.csrf)?;
|
||||
require_access(&state, list_id).await?;
|
||||
let revision = state.db.delete_item(list_id, item_id).await?;
|
||||
state.hub.publish_list_changed(list_id, revision).await;
|
||||
list_fragment_response(&state, &user, list_id).await
|
||||
}
|
||||
|
||||
async fn create_category(
|
||||
State(state): State<AppState>,
|
||||
user: CurrentUser,
|
||||
Path(list_id): Path<i64>,
|
||||
LoggedForm(form): LoggedForm<CategoryForm>,
|
||||
) -> Result<Response, AppError> {
|
||||
verify_csrf(&user, &form.csrf)?;
|
||||
require_access(&state, list_id).await?;
|
||||
let name = form.name.trim().to_owned();
|
||||
if name.is_empty() || name.chars().count() > 60 {
|
||||
return Err(AppError::BadRequest(
|
||||
"Category names must be between 1 and 60 characters.".into(),
|
||||
));
|
||||
}
|
||||
let revision = state.db.create_category(list_id, name).await?;
|
||||
state.hub.publish_list_changed(list_id, revision).await;
|
||||
|
||||
let access = require_access(&state, list_id).await?;
|
||||
let items = state.db.items(list_id).await?;
|
||||
let categories = state.db.categories(list_id).await?;
|
||||
Ok(html_response(views::category_created(
|
||||
&access,
|
||||
&items,
|
||||
&categories,
|
||||
&user.session.csrf_token,
|
||||
)))
|
||||
}
|
||||
|
||||
async fn create_invitation(
|
||||
State(state): State<AppState>,
|
||||
user: CurrentUser,
|
||||
LoggedForm(form): LoggedForm<CsrfForm>,
|
||||
) -> Result<Response, AppError> {
|
||||
verify_csrf(&user, &form.csrf)?;
|
||||
let token = db::new_secret();
|
||||
state
|
||||
.db
|
||||
.create_invitation(user.session.user.id, token.clone())
|
||||
.await?;
|
||||
let url = format!(
|
||||
"{}/invite/{token}",
|
||||
state.public_base_url.trim_end_matches('/')
|
||||
);
|
||||
Ok(html_response(views::invite_result(&url)))
|
||||
}
|
||||
|
||||
async fn invitation_page(
|
||||
State(state): State<AppState>,
|
||||
Path(token): Path<String>,
|
||||
headers: HeaderMap,
|
||||
) -> Result<Response, AppError> {
|
||||
let info = state.db.invitation(token.clone()).await?;
|
||||
if !info {
|
||||
return Err(AppError::NotFound);
|
||||
}
|
||||
let user = optional_user(&state, &headers).await?;
|
||||
Ok(html_response(views::invite_page(
|
||||
user.as_ref().map(|current| ¤t.session.user),
|
||||
&token,
|
||||
None,
|
||||
user.as_ref()
|
||||
.map(|current| current.session.csrf_token.as_str()),
|
||||
)))
|
||||
}
|
||||
|
||||
async fn accept_invitation(
|
||||
State(state): State<AppState>,
|
||||
user: CurrentUser,
|
||||
Path(token): Path<String>,
|
||||
LoggedForm(form): LoggedForm<CsrfForm>,
|
||||
) -> Result<Response, AppError> {
|
||||
verify_csrf(&user, &form.csrf)?;
|
||||
state.db.accept_invitation(token).await?;
|
||||
Ok(Redirect::to("/lists").into_response())
|
||||
}
|
||||
|
||||
async fn list_stream(
|
||||
State(state): State<AppState>,
|
||||
user: CurrentUser,
|
||||
Path(list_id): Path<i64>,
|
||||
websocket: WebSocketUpgrade,
|
||||
) -> Result<Response, AppError> {
|
||||
require_access(&state, list_id).await?;
|
||||
let state_for_socket = state.clone();
|
||||
let user_for_socket = user.clone();
|
||||
Ok(websocket
|
||||
.on_upgrade(move |socket| handle_socket(state_for_socket, user_for_socket, list_id, socket))
|
||||
.into_response())
|
||||
}
|
||||
|
||||
async fn handle_socket(state: AppState, user: CurrentUser, list_id: i64, socket: WebSocket) {
|
||||
let subscription = state
|
||||
.hub
|
||||
.join(
|
||||
list_id,
|
||||
user.session.user.id,
|
||||
user.session.user.display_name.clone(),
|
||||
)
|
||||
.await;
|
||||
let connection_id = subscription.connection_id.clone();
|
||||
let (mut sender, mut receiver) = socket.split();
|
||||
let mut heartbeat = tokio::time::interval(Duration::from_secs(30));
|
||||
heartbeat.tick().await;
|
||||
|
||||
match websocket_snapshot(&state, &user, list_id, &subscription.presence).await {
|
||||
Ok(snapshot) => {
|
||||
if sender.send(Message::Text(snapshot.into())).await.is_err() {
|
||||
state.hub.leave(list_id, &connection_id).await;
|
||||
return;
|
||||
}
|
||||
}
|
||||
Err(error) => {
|
||||
error!(%error, "could not render websocket snapshot");
|
||||
state.hub.leave(list_id, &connection_id).await;
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
let mut events = subscription.receiver;
|
||||
loop {
|
||||
tokio::select! {
|
||||
event = events.recv() => {
|
||||
match event {
|
||||
Ok(HubEvent::ListChanged { list_id: event_list_id, revision }) if event_list_id == list_id => {
|
||||
tracing::debug!(%list_id, revision, "list changed on websocket");
|
||||
match websocket_list_update(&state, &user, list_id).await {
|
||||
Ok(update) => {
|
||||
if sender.send(Message::Text(update.into())).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Err(error) => {
|
||||
error!(%error, "could not render websocket list update");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(HubEvent::PresenceChanged { list_id: event_list_id }) if event_list_id == list_id => {
|
||||
let presence = state.hub.presence(list_id).await;
|
||||
let update = views::presence_panel(&presence, true).into_string();
|
||||
if sender.send(Message::Text(update.into())).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
|
||||
match websocket_snapshot(&state, &user, list_id, &state.hub.presence(list_id).await).await {
|
||||
Ok(snapshot) => {
|
||||
if sender.send(Message::Text(snapshot.into())).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Err(error) => {
|
||||
error!(%error, "could not resync websocket");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
|
||||
Ok(_) => {}
|
||||
}
|
||||
}
|
||||
_ = heartbeat.tick() => {
|
||||
if sender.send(Message::Ping(Vec::new().into())).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
incoming = receiver.next() => {
|
||||
match incoming {
|
||||
Some(Ok(Message::Ping(payload))) => {
|
||||
if sender.send(Message::Pong(payload)).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Some(Ok(Message::Close(_))) | None => break,
|
||||
Some(Ok(_)) => {}
|
||||
Some(Err(_)) => break,
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
state.hub.leave(list_id, &connection_id).await;
|
||||
}
|
||||
|
||||
async fn websocket_snapshot(
|
||||
state: &AppState,
|
||||
user: &CurrentUser,
|
||||
list_id: i64,
|
||||
presence: &[hub::PresenceUser],
|
||||
) -> Result<String, AppError> {
|
||||
let access = require_access(state, list_id).await?;
|
||||
let items = state.db.items(list_id).await?;
|
||||
let categories = state.db.categories(list_id).await?;
|
||||
Ok(
|
||||
views::live_list_fragments(&access, &items, &categories, &user.session.csrf_token)
|
||||
.into_string()
|
||||
+ &views::presence_panel(presence, true).into_string(),
|
||||
)
|
||||
}
|
||||
|
||||
async fn websocket_list_update(
|
||||
state: &AppState,
|
||||
user: &CurrentUser,
|
||||
list_id: i64,
|
||||
) -> Result<String, AppError> {
|
||||
let access = require_access(state, list_id).await?;
|
||||
let items = state.db.items(list_id).await?;
|
||||
let categories = state.db.categories(list_id).await?;
|
||||
Ok(
|
||||
views::live_list_fragments(&access, &items, &categories, &user.session.csrf_token)
|
||||
.into_string(),
|
||||
)
|
||||
}
|
||||
|
||||
async fn list_fragment_response(
|
||||
state: &AppState,
|
||||
user: &CurrentUser,
|
||||
list_id: i64,
|
||||
) -> Result<Response, AppError> {
|
||||
let access = require_access(state, list_id).await?;
|
||||
let items = state.db.items(list_id).await?;
|
||||
let categories = state.db.categories(list_id).await?;
|
||||
Ok(html_response(views::list_items_fragment(
|
||||
&access,
|
||||
&items,
|
||||
&categories,
|
||||
&user.session.csrf_token,
|
||||
false,
|
||||
)))
|
||||
}
|
||||
|
||||
async fn require_access(state: &AppState, list_id: i64) -> Result<db::GroceryList, AppError> {
|
||||
state
|
||||
.db
|
||||
.list_access(list_id)
|
||||
.await?
|
||||
.ok_or(AppError::NotFound)
|
||||
}
|
||||
|
||||
async fn optional_user(
|
||||
state: &AppState,
|
||||
headers: &HeaderMap,
|
||||
) -> Result<Option<CurrentUser>, AppError> {
|
||||
let Some(session_token) = cookie_value(headers, "session") else {
|
||||
return Ok(None);
|
||||
};
|
||||
Ok(state
|
||||
.db
|
||||
.session_user(session_token.clone())
|
||||
.await?
|
||||
.map(|session| CurrentUser {
|
||||
session_token,
|
||||
session,
|
||||
}))
|
||||
}
|
||||
|
||||
fn verify_csrf(user: &CurrentUser, token: &str) -> Result<(), AppError> {
|
||||
if token.is_empty() || token != user.session.csrf_token {
|
||||
return Err(AppError::BadRequest(
|
||||
"Your form has expired. Refresh and try again.".into(),
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn parse_category_id(category_id: Option<String>) -> Option<i64> {
|
||||
category_id
|
||||
.filter(|category_id| !category_id.trim().is_empty())
|
||||
.and_then(|category_id| category_id.trim().parse().ok())
|
||||
}
|
||||
|
||||
async fn can_register(state: &AppState, invite: Option<&str>) -> Result<bool, AppError> {
|
||||
if state.registration_mode == RegistrationMode::Open {
|
||||
return Ok(true);
|
||||
}
|
||||
if !state.db.has_users().await? {
|
||||
return Ok(true);
|
||||
}
|
||||
let Some(invite) = invite.filter(|invite| !invite.is_empty()) else {
|
||||
return Ok(false);
|
||||
};
|
||||
Ok(state.db.invitation(invite.to_owned()).await?)
|
||||
}
|
||||
|
||||
fn hash_password(password: &str) -> Result<String, String> {
|
||||
let salt = SaltString::generate(&mut OsRng);
|
||||
Argon2::default()
|
||||
.hash_password(password.as_bytes(), &salt)
|
||||
.map(|hash| hash.to_string())
|
||||
.map_err(|error| error.to_string())
|
||||
}
|
||||
|
||||
fn verify_password(password: &str, encoded_hash: &str) -> Result<bool, String> {
|
||||
let hash = PasswordHash::new(encoded_hash).map_err(|error| error.to_string())?;
|
||||
Ok(Argon2::default()
|
||||
.verify_password(password.as_bytes(), &hash)
|
||||
.is_ok())
|
||||
}
|
||||
|
||||
fn html_response(markup: maud::Markup) -> Response {
|
||||
Html(markup.into_string()).into_response()
|
||||
}
|
||||
|
||||
fn status_html_response(status: StatusCode, markup: maud::Markup) -> Response {
|
||||
(status, Html(markup.into_string())).into_response()
|
||||
}
|
||||
|
||||
fn cookie_value(headers: &HeaderMap, name: &str) -> Option<String> {
|
||||
headers
|
||||
.get(header::COOKIE)?
|
||||
.to_str()
|
||||
.ok()?
|
||||
.split(';')
|
||||
.map(str::trim)
|
||||
.find_map(|cookie| {
|
||||
let (key, value) = cookie.split_once('=')?;
|
||||
(key == name).then(|| value.to_owned())
|
||||
})
|
||||
}
|
||||
|
||||
fn set_session_cookie(response: &mut Response, token: &str, secure: bool) {
|
||||
let secure_attribute = if secure { "; Secure" } else { "" };
|
||||
let cookie = format!(
|
||||
"session={token}; Path=/; HttpOnly; SameSite=Lax; Max-Age=2592000{secure_attribute}"
|
||||
);
|
||||
response.headers_mut().append(
|
||||
header::SET_COOKIE,
|
||||
HeaderValue::from_str(&cookie).expect("session cookie is valid"),
|
||||
);
|
||||
}
|
||||
|
||||
fn clear_session_cookie(response: &mut Response, secure: bool) {
|
||||
let secure_attribute = if secure { "; Secure" } else { "" };
|
||||
let cookie = format!("session=; Path=/; HttpOnly; SameSite=Lax; Max-Age=0{secure_attribute}");
|
||||
response.headers_mut().append(
|
||||
header::SET_COOKIE,
|
||||
HeaderValue::from_str(&cookie).expect("session cookie is valid"),
|
||||
);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user