Compare commits
8 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 489d44fa8a | |||
| 83dfbf560e | |||
| d9d7a25eca | |||
| 45a4c1eceb | |||
| 1d1de4bcec | |||
| 2564c33748 | |||
| 9c7cb1c9b8 | |||
| bc4de69ab9 |
@@ -0,0 +1,5 @@
|
||||
target/
|
||||
data/
|
||||
*.db
|
||||
.git/
|
||||
tests/
|
||||
Generated
+2609
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,25 @@
|
||||
[package]
|
||||
name = "ares-server"
|
||||
version = "0.1.0"
|
||||
edition = "2021"
|
||||
publish = false
|
||||
|
||||
[dependencies]
|
||||
axum = { version = "0.8", features = ["ws"] }
|
||||
tokio = { version = "1", features = ["full", "macros", "rt-multi-thread"] }
|
||||
serde = { version = "1", features = ["derive"] }
|
||||
serde_json = "1"
|
||||
tracing = "0.1"
|
||||
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
|
||||
tower-http = { version = "0.6", features = ["cors", "trace"] }
|
||||
rusqlite = { version = "0.32", features = ["bundled"] }
|
||||
thiserror = "2"
|
||||
chrono = { version = "0.4", default-features = false, features = ["clock"] }
|
||||
futures-util = "0.3"
|
||||
tokio-stream = { version = "0.1", features = ["sync"] }
|
||||
|
||||
[dev-dependencies]
|
||||
axum-test = "21"
|
||||
tempfile = "3"
|
||||
tokio-tungstenite = "0.24"
|
||||
reqwest = { version = "0.12", features = ["json"] }
|
||||
@@ -0,0 +1,18 @@
|
||||
FROM rust:1.85-alpine AS builder
|
||||
WORKDIR /build
|
||||
RUN apk add --no-cache musl-dev build-base
|
||||
COPY Cargo.toml ./
|
||||
COPY Cargo.lock ./
|
||||
COPY src ./src
|
||||
RUN cargo build --release
|
||||
|
||||
FROM alpine:3.20
|
||||
RUN apk add --no-cache ca-certificates tzdata
|
||||
WORKDIR /app
|
||||
COPY --from=builder /build/target/release/ares-server /app/ares-server
|
||||
ENV ARS_NEXT_HOST=0.0.0.0
|
||||
ENV ARS_NEXT_PORT=3000
|
||||
ENV ARS_NEXT_DB=/data/aresnext.db
|
||||
EXPOSE 3000
|
||||
VOLUME ["/data"]
|
||||
CMD ["/app/ares-server"]
|
||||
@@ -0,0 +1,27 @@
|
||||
use axum::{extract::State, Json};
|
||||
use crate::error::{AppError, AppResult};
|
||||
use crate::models::AnnounceRequest;
|
||||
use crate::state::SharedState;
|
||||
use crate::validate;
|
||||
|
||||
pub async fn announce(
|
||||
State(state): State<SharedState>,
|
||||
Json(req): Json<AnnounceRequest>,
|
||||
) -> AppResult<Json<serde_json::Value>> {
|
||||
validate::validate_announce(&req)?;
|
||||
|
||||
if !state.rate.lock().unwrap().check(&req.peer_id) {
|
||||
return Err(AppError::BadRequest("rate limit exceeded".into()));
|
||||
}
|
||||
|
||||
let db = state.db.lock().unwrap();
|
||||
let id = crate::store::insert_file(&db, &req).map_err(|e| match e {
|
||||
rusqlite::Error::SqliteFailure(err, _)
|
||||
if err.code == rusqlite::ErrorCode::ConstraintViolation =>
|
||||
{
|
||||
AppError::BadRequest("duplicate file (hash already announced)".into())
|
||||
}
|
||||
other => AppError::Db(other),
|
||||
})?;
|
||||
Ok(Json(serde_json::json!({ "file_id": id, "ok": true })))
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
use axum::extract::{Path, State};
|
||||
use axum::Json;
|
||||
use crate::error::{AppError, AppResult};
|
||||
use crate::models::FileRecord;
|
||||
use crate::state::SharedState;
|
||||
|
||||
pub async fn get_file(
|
||||
State(state): State<SharedState>,
|
||||
Path(file_id): Path<String>,
|
||||
) -> AppResult<Json<FileRecord>> {
|
||||
let db = state.db.lock().unwrap();
|
||||
crate::store::get_file(&db, &file_id)
|
||||
.map(Json)
|
||||
.map_err(|_| AppError::NotFound("file not found".into()))
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
use axum::Json;
|
||||
use serde_json::{json, Value};
|
||||
|
||||
pub async fn health() -> Json<Value> {
|
||||
Json(json!({
|
||||
"status": "ok",
|
||||
"version": env!("CARGO_PKG_VERSION"),
|
||||
"service": "ares-next",
|
||||
}))
|
||||
}
|
||||
@@ -0,0 +1,27 @@
|
||||
pub mod announce;
|
||||
pub mod files;
|
||||
pub mod health;
|
||||
pub mod peers;
|
||||
pub mod search;
|
||||
pub mod stats;
|
||||
pub mod ws;
|
||||
|
||||
use axum::{routing::{get, post}, Router};
|
||||
use tower_http::cors::CorsLayer;
|
||||
use crate::state::SharedState;
|
||||
|
||||
pub fn router(state: SharedState) -> Router {
|
||||
let api = Router::new()
|
||||
.route("/health", get(health::health))
|
||||
.route("/search", get(search::search))
|
||||
.route("/files/{file_id}", get(files::get_file))
|
||||
.route("/announce", post(announce::announce))
|
||||
.route("/peers/{file_id}", get(peers::peers))
|
||||
.route("/stats", get(stats::stats))
|
||||
.route("/tracker/ws", get(ws::ws_handler))
|
||||
.with_state(state);
|
||||
|
||||
Router::new()
|
||||
.nest("/api/v1", api)
|
||||
.layer(CorsLayer::permissive())
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
use axum::extract::{Path, State};
|
||||
use axum::Json;
|
||||
use crate::error::AppResult;
|
||||
use crate::state::SharedState;
|
||||
|
||||
pub async fn peers(
|
||||
State(state): State<SharedState>,
|
||||
Path(file_id): Path<String>,
|
||||
) -> AppResult<Json<serde_json::Value>> {
|
||||
let t = state.tracker.lock().unwrap();
|
||||
let peers = t.peers_of(&file_id);
|
||||
Ok(Json(serde_json::json!({ "file_id": file_id, "peers": peers })))
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
use axum::extract::{Query, State};
|
||||
use axum::Json;
|
||||
use crate::error::AppResult;
|
||||
use crate::models::{FileRecord, SearchQuery};
|
||||
use crate::state::SharedState;
|
||||
|
||||
pub async fn search(
|
||||
State(state): State<SharedState>,
|
||||
Query(q): Query<SearchQuery>,
|
||||
) -> AppResult<Json<Vec<FileRecord>>> {
|
||||
let db = state.db.lock().unwrap();
|
||||
let limit = q.limit.unwrap_or(state.config.max_results).clamp(1, 100);
|
||||
let offset = q.offset.unwrap_or(0).max(0);
|
||||
let rows = crate::store::search_files(
|
||||
&db,
|
||||
q.q.as_deref().unwrap_or(""),
|
||||
q.ftype.as_deref(),
|
||||
limit,
|
||||
offset,
|
||||
)?;
|
||||
Ok(Json(rows))
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
use axum::extract::State;
|
||||
use axum::Json;
|
||||
use crate::error::AppResult;
|
||||
use crate::state::SharedState;
|
||||
|
||||
pub async fn stats(State(state): State<SharedState>) -> AppResult<Json<serde_json::Value>> {
|
||||
let db = state.db.lock().unwrap();
|
||||
let files: i64 = db.query_row("SELECT count(*) FROM files", [], |r| r.get(0))?;
|
||||
let bytes: i64 = db.query_row("SELECT COALESCE(SUM(size_bytes),0) FROM files", [], |r| r.get(0))?;
|
||||
let t = state.tracker.lock().unwrap();
|
||||
let online_peers = t.online_total();
|
||||
Ok(Json(serde_json::json!({
|
||||
"files": files,
|
||||
"total_bytes": bytes,
|
||||
"online_peers": online_peers,
|
||||
"version": env!("CARGO_PKG_VERSION"),
|
||||
})))
|
||||
}
|
||||
@@ -0,0 +1,94 @@
|
||||
use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
|
||||
use axum::extract::State;
|
||||
use axum::response::Response;
|
||||
use futures_util::{SinkExt, StreamExt};
|
||||
use tokio::sync::mpsc;
|
||||
use tokio_stream::wrappers::UnboundedReceiverStream;
|
||||
use tracing::{debug, warn};
|
||||
|
||||
use crate::state::SharedState;
|
||||
|
||||
#[derive(serde::Deserialize, Debug)]
|
||||
struct RegisterMsg {
|
||||
peer_id: String,
|
||||
file_id: String,
|
||||
}
|
||||
|
||||
pub async fn ws_handler(
|
||||
ws: WebSocketUpgrade,
|
||||
State(state): State<SharedState>,
|
||||
) -> Response {
|
||||
ws.on_upgrade(move |socket| handle_socket(socket, state))
|
||||
}
|
||||
|
||||
async fn handle_socket(socket: WebSocket, state: SharedState) {
|
||||
let (mut sender, mut receiver) = socket.split();
|
||||
let (tx, rx) = mpsc::unbounded_channel::<Message>();
|
||||
let mut recv_stream = UnboundedReceiverStream::new(rx);
|
||||
|
||||
let registration = wait_for_registration(&mut receiver).await;
|
||||
let (file_id, peer_id) = match registration {
|
||||
Some((file_id, peer_id)) => {
|
||||
let mut t = state.tracker.lock().unwrap();
|
||||
t.register(file_id.clone(), peer_id.clone(), tx.clone());
|
||||
(file_id, peer_id)
|
||||
}
|
||||
None => return,
|
||||
};
|
||||
|
||||
// notifica os demais pares do mesmo arquivo (exceto o próprio)
|
||||
{
|
||||
let t = state.tracker.lock().unwrap();
|
||||
t.broadcast_except(
|
||||
&file_id,
|
||||
&peer_id,
|
||||
&format!(
|
||||
r#"{{"type":"peer_online","file_id":"{}","peer_id":"{}"}}"#,
|
||||
file_id, peer_id
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
msg = receiver.next() => {
|
||||
match msg {
|
||||
Some(Ok(Message::Close(_))) | None => break,
|
||||
Some(Ok(Message::Ping(p))) => {
|
||||
if sender.send(Message::Pong(p)).await.is_err() { break; }
|
||||
}
|
||||
Some(Ok(_)) => {} // ignoramos frames não relevantes
|
||||
Some(Err(e)) => {
|
||||
warn!("ws error: {e}");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
msg = recv_stream.next() => {
|
||||
if let Some(m) = msg {
|
||||
if sender.send(m).await.is_err() { break; }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let mut t = state.tracker.lock().unwrap();
|
||||
t.unregister(&file_id, &peer_id);
|
||||
debug!("peer {peer_id} removed from {file_id}");
|
||||
}
|
||||
|
||||
/// Awaits the first text frame carrying `peer_id` + `file_id`.
|
||||
async fn wait_for_registration(receiver: &mut futures_util::stream::SplitStream<WebSocket>) -> Option<(String, String)> {
|
||||
loop {
|
||||
match receiver.next().await {
|
||||
Some(Ok(Message::Text(text))) => {
|
||||
if let Ok(reg) = serde_json::from_str::<RegisterMsg>(&text) {
|
||||
return Some((reg.file_id, reg.peer_id));
|
||||
}
|
||||
}
|
||||
Some(Ok(Message::Close(_))) | None => return None,
|
||||
Some(Err(_)) => return None,
|
||||
_ => continue,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
use std::env;
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct Config {
|
||||
pub host: String,
|
||||
pub port: u16,
|
||||
pub db_path: String,
|
||||
pub max_body_mb: usize,
|
||||
pub max_results: i64,
|
||||
}
|
||||
|
||||
impl Default for Config {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
host: "127.0.0.1".into(),
|
||||
port: 3000,
|
||||
db_path: "data/aresnext.db".into(),
|
||||
max_body_mb: 2,
|
||||
max_results: 50,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Config {
|
||||
pub fn from_env() -> Self {
|
||||
let def = Self::default();
|
||||
Self {
|
||||
host: env::var("ARS_NEXT_HOST").unwrap_or(def.host),
|
||||
port: env::var("ARS_NEXT_PORT")
|
||||
.ok()
|
||||
.and_then(|s| s.parse().ok())
|
||||
.unwrap_or(def.port),
|
||||
db_path: env::var("ARS_NEXT_DB").unwrap_or(def.db_path),
|
||||
max_body_mb: env::var("ARS_NEXT_MAX_BODY_MB")
|
||||
.ok()
|
||||
.and_then(|s| s.parse().ok())
|
||||
.unwrap_or(def.max_body_mb),
|
||||
max_results: env::var("ARS_NEXT_MAX_RESULTS")
|
||||
.ok()
|
||||
.and_then(|s| s.parse().ok())
|
||||
.unwrap_or(def.max_results),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn max_body_bytes(&self) -> usize {
|
||||
self.max_body_mb * 1024 * 1024
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
use axum::{http::StatusCode, response::{IntoResponse, Response}, Json};
|
||||
use serde_json::json;
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum AppError {
|
||||
#[error("bad request: {0}")]
|
||||
BadRequest(String),
|
||||
#[error("not found: {0}")]
|
||||
NotFound(String),
|
||||
#[error("database error: {0}")]
|
||||
Db(#[from] rusqlite::Error),
|
||||
#[error("internal: {0}")]
|
||||
Internal(String),
|
||||
}
|
||||
|
||||
impl AppError {
|
||||
fn status(&self) -> StatusCode {
|
||||
match self {
|
||||
Self::BadRequest(_) => StatusCode::BAD_REQUEST,
|
||||
Self::NotFound(_) => StatusCode::NOT_FOUND,
|
||||
Self::Db(_) | Self::Internal(_) => StatusCode::INTERNAL_SERVER_ERROR,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl IntoResponse for AppError {
|
||||
fn into_response(self) -> Response {
|
||||
let body = json!({ "error": self.to_string() });
|
||||
(self.status(), Json(body)).into_response()
|
||||
}
|
||||
}
|
||||
|
||||
pub type AppResult<T> = Result<T, AppError>;
|
||||
@@ -0,0 +1,23 @@
|
||||
pub mod api;
|
||||
pub mod config;
|
||||
pub mod error;
|
||||
pub mod models;
|
||||
pub mod rate;
|
||||
pub mod search;
|
||||
pub mod state;
|
||||
pub mod store;
|
||||
pub mod validate;
|
||||
pub mod ws;
|
||||
|
||||
use axum::Router;
|
||||
use crate::config::Config;
|
||||
use crate::state::{AppState, SharedState};
|
||||
|
||||
pub async fn build_app(cfg: Config) -> SharedState {
|
||||
let conn = store::open(&cfg).expect("open database");
|
||||
AppState::new(cfg, conn)
|
||||
}
|
||||
|
||||
pub fn router(state: SharedState) -> Router {
|
||||
api::router(state)
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
use ares_server::config::Config;
|
||||
use ares_server::{build_app, router};
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() {
|
||||
tracing_subscriber::fmt()
|
||||
.with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
|
||||
.init();
|
||||
|
||||
let cfg = Config::from_env();
|
||||
let addr = format!("{}:{}", cfg.host, cfg.port);
|
||||
let state = build_app(cfg).await;
|
||||
let app = router(state);
|
||||
|
||||
let listener = tokio::net::TcpListener::bind(&addr)
|
||||
.await
|
||||
.expect("bind listener");
|
||||
tracing::info!("AresNext listening on http://{addr}");
|
||||
axum::serve(listener, app).await.expect("server");
|
||||
}
|
||||
@@ -0,0 +1,61 @@
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
pub const FILE_TYPES: [&str; 5] = ["audio", "video", "document", "software", "other"];
|
||||
|
||||
#[derive(Serialize, Clone, Debug)]
|
||||
pub struct FileRecord {
|
||||
pub id: String,
|
||||
pub peer_id: String,
|
||||
pub name: String,
|
||||
pub description: String,
|
||||
pub ftype: String,
|
||||
pub extension: String,
|
||||
pub size_bytes: i64,
|
||||
pub hash: String,
|
||||
pub license: String,
|
||||
pub tags: Vec<String>,
|
||||
pub created_at: i64,
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Debug)]
|
||||
pub struct AnnounceRequest {
|
||||
pub peer_id: String,
|
||||
pub name: String,
|
||||
pub description: String,
|
||||
#[serde(rename = "type")]
|
||||
pub ftype: String,
|
||||
pub extension: String,
|
||||
pub size_bytes: i64,
|
||||
pub hash: String,
|
||||
pub license: String,
|
||||
#[serde(default)]
|
||||
pub tags: Vec<String>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Debug)]
|
||||
pub struct SearchQuery {
|
||||
pub q: Option<String>,
|
||||
pub ftype: Option<String>,
|
||||
pub sort: Option<String>,
|
||||
pub limit: Option<i64>,
|
||||
pub offset: Option<i64>,
|
||||
}
|
||||
|
||||
impl FileRecord {
|
||||
pub(crate) fn from_row(r: &rusqlite::Row) -> rusqlite::Result<Self> {
|
||||
let tags: String = r.get(9)?;
|
||||
Ok(Self {
|
||||
id: r.get(0)?,
|
||||
peer_id: r.get(1)?,
|
||||
name: r.get(2)?,
|
||||
description: r.get(3)?,
|
||||
ftype: r.get(4)?,
|
||||
extension: r.get(5)?,
|
||||
size_bytes: r.get(6)?,
|
||||
hash: r.get(7)?,
|
||||
license: r.get(8)?,
|
||||
tags: serde_json::from_str(&tags).unwrap_or_default(),
|
||||
created_at: r.get(10)?,
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
use std::collections::HashMap;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
pub struct RateLimiter {
|
||||
window: Duration,
|
||||
max: u32,
|
||||
hits: HashMap<String, (Instant, u32)>,
|
||||
}
|
||||
|
||||
impl RateLimiter {
|
||||
pub fn new(per_second: u32) -> Self {
|
||||
Self {
|
||||
window: Duration::from_secs(1),
|
||||
max: per_second,
|
||||
hits: HashMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn check(&mut self, key: &str) -> bool {
|
||||
let now = Instant::now();
|
||||
let entry = self.hits.entry(key.to_string()).or_insert((now, 0));
|
||||
if now.duration_since(entry.0) > self.window {
|
||||
*entry = (now, 0);
|
||||
}
|
||||
entry.1 += 1;
|
||||
entry.1 <= self.max
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn allows_until_limit() {
|
||||
let mut rl = RateLimiter::new(2);
|
||||
assert!(rl.check("k"));
|
||||
assert!(rl.check("k"));
|
||||
assert!(!rl.check("k"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn keys_are_isolated() {
|
||||
let mut rl = RateLimiter::new(1);
|
||||
assert!(rl.check("a"));
|
||||
assert!(rl.check("b"));
|
||||
assert!(!rl.check("a"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn resets_after_window() {
|
||||
let mut rl = RateLimiter::new(1);
|
||||
assert!(rl.check("k"));
|
||||
assert!(!rl.check("k"));
|
||||
rl.hits.get_mut("k").unwrap().0 = Instant::now() - Duration::from_secs(2);
|
||||
assert!(rl.check("k"));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
pub fn fts_query(q: &str) -> Option<String> {
|
||||
let terms: Vec<String> = q
|
||||
.split_whitespace()
|
||||
.map(|t| t.trim_matches(['"', '\'', '*', '(', ')', ':', '[', ']']))
|
||||
.filter(|t| !t.is_empty())
|
||||
.map(|t| format!("\"{}\"", t))
|
||||
.collect();
|
||||
if terms.is_empty() {
|
||||
None
|
||||
} else {
|
||||
Some(terms.join(" AND "))
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,27 @@
|
||||
use std::sync::{Arc, Mutex};
|
||||
use rusqlite::Connection;
|
||||
use crate::config::Config;
|
||||
use crate::rate::RateLimiter;
|
||||
use crate::ws::tracker::PeerTracker;
|
||||
|
||||
pub type DbConn = Mutex<Connection>;
|
||||
|
||||
pub struct AppState {
|
||||
pub config: Config,
|
||||
pub db: DbConn,
|
||||
pub tracker: Mutex<PeerTracker>,
|
||||
pub rate: Mutex<RateLimiter>,
|
||||
}
|
||||
|
||||
impl AppState {
|
||||
pub fn new(config: Config, db: Connection) -> Arc<Self> {
|
||||
Arc::new(Self {
|
||||
config,
|
||||
db: Mutex::new(db),
|
||||
tracker: Mutex::new(PeerTracker::default()),
|
||||
rate: Mutex::new(RateLimiter::new(10)),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
pub type SharedState = Arc<AppState>;
|
||||
@@ -0,0 +1,116 @@
|
||||
use rusqlite::{params, Connection};
|
||||
use crate::config::Config;
|
||||
use crate::error::AppResult;
|
||||
use crate::models::{AnnounceRequest, FileRecord};
|
||||
|
||||
pub fn open(cfg: &Config) -> AppResult<Connection> {
|
||||
if let Some(parent) = std::path::Path::new(&cfg.db_path).parent() {
|
||||
std::fs::create_dir_all(parent).ok();
|
||||
}
|
||||
let db = Connection::open(&cfg.db_path)?;
|
||||
db.pragma_update(None, "journal_mode", "WAL")?;
|
||||
db.pragma_update(None, "synchronous", "NORMAL")?;
|
||||
migrate(&db)?;
|
||||
Ok(db)
|
||||
}
|
||||
|
||||
pub fn migrate(db: &Connection) -> rusqlite::Result<()> {
|
||||
db.execute_batch(
|
||||
r#"
|
||||
CREATE TABLE IF NOT EXISTS files (
|
||||
id TEXT PRIMARY KEY,
|
||||
peer_id TEXT NOT NULL,
|
||||
name TEXT NOT NULL,
|
||||
description TEXT NOT NULL DEFAULT '',
|
||||
ftype TEXT NOT NULL,
|
||||
extension TEXT NOT NULL DEFAULT '',
|
||||
size_bytes INTEGER NOT NULL,
|
||||
hash TEXT NOT NULL UNIQUE,
|
||||
license TEXT NOT NULL DEFAULT 'authored',
|
||||
tags TEXT NOT NULL DEFAULT '[]',
|
||||
created_at INTEGER NOT NULL
|
||||
);
|
||||
|
||||
CREATE VIRTUAL TABLE IF NOT EXISTS files_fts USING fts5(
|
||||
name, description, tags,
|
||||
content='files', content_rowid='rowid',
|
||||
tokenize='unicode61'
|
||||
);
|
||||
|
||||
CREATE TRIGGER IF NOT EXISTS files_ai AFTER INSERT ON files BEGIN
|
||||
INSERT INTO files_fts(rowid, name, description, tags)
|
||||
VALUES (new.rowid, new.name, new.description, new.tags);
|
||||
END;
|
||||
|
||||
CREATE TRIGGER IF NOT EXISTS files_ad AFTER DELETE ON files BEGIN
|
||||
INSERT INTO files_fts(files_fts, rowid, name, description, tags)
|
||||
VALUES ('delete', old.rowid, old.name, old.description, old.tags);
|
||||
END;
|
||||
"#,
|
||||
)
|
||||
}
|
||||
|
||||
pub fn insert_file(db: &Connection, req: &AnnounceRequest) -> rusqlite::Result<String> {
|
||||
let id = format!("{}::{}", req.peer_id, &req.hash[..12]);
|
||||
let tags_json = serde_json::to_string(&req.tags).unwrap_or_else(|_| "[]".into());
|
||||
db.execute(
|
||||
"INSERT INTO files (id, peer_id, name, description, ftype, extension, size_bytes, hash, license, tags, created_at)
|
||||
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)",
|
||||
params![
|
||||
id, req.peer_id, req.name, req.description, req.ftype, req.extension,
|
||||
req.size_bytes, req.hash, req.license, tags_json,
|
||||
chrono::Utc::now().timestamp(),
|
||||
],
|
||||
)?;
|
||||
Ok(id)
|
||||
}
|
||||
|
||||
pub fn get_file(db: &Connection, file_id: &str) -> rusqlite::Result<FileRecord> {
|
||||
db.query_row(
|
||||
"SELECT id, peer_id, name, description, ftype, extension, size_bytes, hash, license, tags, created_at
|
||||
FROM files WHERE id = ?1",
|
||||
[file_id],
|
||||
FileRecord::from_row,
|
||||
)
|
||||
}
|
||||
|
||||
pub fn search_files(
|
||||
db: &Connection,
|
||||
q: &str,
|
||||
ftype: Option<&str>,
|
||||
limit: i64,
|
||||
offset: i64,
|
||||
) -> rusqlite::Result<Vec<FileRecord>> {
|
||||
let mut sql = String::from(
|
||||
"SELECT f.id, f.peer_id, f.name, f.description, f.ftype, f.extension,
|
||||
f.size_bytes, f.hash, f.license, f.tags, f.created_at
|
||||
FROM files f",
|
||||
);
|
||||
let mut params: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
|
||||
let mut clauses: Vec<String> = Vec::new();
|
||||
|
||||
if let Some(query) = crate::search::fts_query(q) {
|
||||
sql.push_str(" JOIN files_fts ON files_fts.rowid = f.rowid");
|
||||
clauses.push("files_fts MATCH ?".to_string());
|
||||
params.push(Box::new(query));
|
||||
}
|
||||
if let Some(ft) = ftype {
|
||||
params.push(Box::new(ft.to_string()));
|
||||
clauses.push(format!("f.ftype = ?{}", params.len()));
|
||||
}
|
||||
if !clauses.is_empty() {
|
||||
sql.push_str(" WHERE ");
|
||||
sql.push_str(&clauses.join(" AND "));
|
||||
}
|
||||
params.push(Box::new(limit));
|
||||
sql.push_str(&format!(" ORDER BY f.created_at DESC LIMIT ?{}", params.len()));
|
||||
params.push(Box::new(offset));
|
||||
sql.push_str(&format!(" OFFSET ?{}", params.len()));
|
||||
|
||||
let mut stmt = db.prepare(&sql)?;
|
||||
let rows = stmt.query_map(
|
||||
rusqlite::params_from_iter(params.iter().map(|p| p.as_ref())),
|
||||
FileRecord::from_row,
|
||||
)?;
|
||||
rows.collect()
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
use crate::error::AppError;
|
||||
use crate::models::{AnnounceRequest, FILE_TYPES};
|
||||
|
||||
pub const BANNED_EXT: [&str; 12] = [
|
||||
"exe", "bat", "cmd", "msi", "scr", "ps1", "vbs", "js", "jar", "apk", "dll", "sh",
|
||||
];
|
||||
|
||||
const MAX_NAME: usize = 255;
|
||||
const MAX_TAGS: usize = 20;
|
||||
const MAX_TAG_LEN: usize = 32;
|
||||
const MAX_PEER_ID: usize = 128;
|
||||
|
||||
pub fn validate_announce(req: &AnnounceRequest) -> Result<(), AppError> {
|
||||
if req.name.trim().is_empty() || req.name.len() > MAX_NAME {
|
||||
return Err(AppError::BadRequest("invalid name".into()));
|
||||
}
|
||||
if !FILE_TYPES.contains(&req.ftype.as_str()) {
|
||||
return Err(AppError::BadRequest(format!("unknown type '{}'", req.ftype)));
|
||||
}
|
||||
if !req.extension.is_empty() {
|
||||
let ext = req.extension.to_lowercase().trim_start_matches('.').to_string();
|
||||
if BANNED_EXT.contains(&ext.as_str()) {
|
||||
return Err(AppError::BadRequest(format!("extension '{ext}' not allowed")));
|
||||
}
|
||||
}
|
||||
if req.size_bytes < 0 {
|
||||
return Err(AppError::BadRequest("invalid size".into()));
|
||||
}
|
||||
if !is_64_hex(&req.hash) {
|
||||
return Err(AppError::BadRequest("hash must be 64 hex chars".into()));
|
||||
}
|
||||
if req.peer_id.trim().is_empty() || req.peer_id.len() > MAX_PEER_ID {
|
||||
return Err(AppError::BadRequest("invalid peer_id".into()));
|
||||
}
|
||||
if req.tags.len() > MAX_TAGS {
|
||||
return Err(AppError::BadRequest("too many tags".into()));
|
||||
}
|
||||
for t in &req.tags {
|
||||
if t.len() > MAX_TAG_LEN {
|
||||
return Err(AppError::BadRequest("tag too long".into()));
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn is_64_hex(s: &str) -> bool {
|
||||
s.len() == 64 && s.chars().all(|c| c.is_ascii_hexdigit())
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
pub mod tracker;
|
||||
@@ -0,0 +1,50 @@
|
||||
use std::collections::HashMap;
|
||||
use tokio::sync::mpsc;
|
||||
use axum::extract::ws::Message;
|
||||
|
||||
pub type WsSender = mpsc::UnboundedSender<Message>;
|
||||
|
||||
#[derive(Default)]
|
||||
pub struct PeerTracker {
|
||||
pub by_file: HashMap<String, HashMap<String, WsSender>>,
|
||||
}
|
||||
|
||||
impl PeerTracker {
|
||||
pub fn register(&mut self, file_id: String, peer_id: String, tx: WsSender) {
|
||||
self.by_file.entry(file_id).or_default().insert(peer_id, tx);
|
||||
}
|
||||
|
||||
pub fn unregister(&mut self, file_id: &str, peer_id: &str) {
|
||||
if let Some(peers) = self.by_file.get_mut(file_id) {
|
||||
peers.remove(peer_id);
|
||||
if peers.is_empty() {
|
||||
self.by_file.remove(file_id);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn peers_of(&self, file_id: &str) -> Vec<String> {
|
||||
self.by_file
|
||||
.get(file_id)
|
||||
.map(|peers| peers.keys().cloned().collect())
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
pub fn online_total(&self) -> usize {
|
||||
self.by_file
|
||||
.values()
|
||||
.flat_map(|peers| peers.keys())
|
||||
.collect::<std::collections::HashSet<_>>()
|
||||
.len()
|
||||
}
|
||||
|
||||
pub fn broadcast_except(&self, file_id: &str, except_peer: &str, payload: &str) {
|
||||
if let Some(peers) = self.by_file.get(file_id) {
|
||||
for (pid, tx) in peers {
|
||||
if pid != except_peer {
|
||||
let _ = tx.send(Message::Text(payload.to_string().into()));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,107 @@
|
||||
mod common;
|
||||
|
||||
use common::test_server;
|
||||
use serde_json::json;
|
||||
|
||||
#[tokio::test]
|
||||
async fn announce_valid_ok() {
|
||||
let server = test_server();
|
||||
let res = server.post("/api/v1/announce")
|
||||
.json(&json!({
|
||||
"peer_id": "peer123",
|
||||
"name": "minha-musica.ogg",
|
||||
"description": "minha composicao",
|
||||
"type": "audio",
|
||||
"extension": "ogg",
|
||||
"size_bytes": 1024,
|
||||
"hash": "ab".repeat(32),
|
||||
"license": "authored",
|
||||
"tags": ["musica"]
|
||||
}))
|
||||
.await;
|
||||
assert_eq!(res.status_code(), 200);
|
||||
assert_eq!(res.json::<serde_json::Value>()["ok"], true);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn announce_invalid_name_rejected() {
|
||||
let server = test_server();
|
||||
let res = server.post("/api/v1/announce")
|
||||
.json(&json!({
|
||||
"peer_id": "peer123",
|
||||
"name": "",
|
||||
"description": "",
|
||||
"type": "audio",
|
||||
"extension": "ogg",
|
||||
"size_bytes": 10,
|
||||
"hash": "ab".repeat(32),
|
||||
"license": "authored"
|
||||
}))
|
||||
.await;
|
||||
assert_eq!(res.status_code(), 400);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn announce_bad_hash_rejected() {
|
||||
let server = test_server();
|
||||
let res = server.post("/api/v1/announce")
|
||||
.json(&json!({
|
||||
"peer_id": "peer123",
|
||||
"name": "arquivo.ogg",
|
||||
"description": "",
|
||||
"type": "audio",
|
||||
"extension": "ogg",
|
||||
"size_bytes": 10,
|
||||
"hash": "ab",
|
||||
"license": "authored"
|
||||
}))
|
||||
.await;
|
||||
assert_eq!(res.status_code(), 400);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn announce_duplicate_hash_rejected() {
|
||||
let server = test_server();
|
||||
let body = json!({
|
||||
"peer_id": "peer123",
|
||||
"name": "musica.ogg",
|
||||
"description": "",
|
||||
"type": "audio",
|
||||
"extension": "ogg",
|
||||
"size_bytes": 100,
|
||||
"hash": "ab".repeat(32),
|
||||
"license": "authored"
|
||||
});
|
||||
server.post("/api/v1/announce").json(&body).await;
|
||||
let res = server.post("/api/v1/announce").json(&body).await;
|
||||
assert_eq!(res.status_code(), 400);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn get_file_roundtrip() {
|
||||
let server = test_server();
|
||||
let hash = "cd".repeat(32);
|
||||
server.post("/api/v1/announce")
|
||||
.json(&json!({
|
||||
"peer_id": "peer1",
|
||||
"name": "doc.pdf",
|
||||
"description": "",
|
||||
"type": "document",
|
||||
"extension": "pdf",
|
||||
"size_bytes": 100,
|
||||
"hash": hash,
|
||||
"license": "authored"
|
||||
}))
|
||||
.await;
|
||||
let file_id = format!("peer1::{}", &hash[..12]);
|
||||
let res = server.get(&format!("/api/v1/files/{file_id}")).await;
|
||||
assert_eq!(res.status_code(), 200);
|
||||
assert_eq!(res.json::<serde_json::Value>()["name"], "doc.pdf");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn get_file_missing_returns_404() {
|
||||
let server = test_server();
|
||||
let res = server.get("/api/v1/files/nao-existe").await;
|
||||
assert_eq!(res.status_code(), 404);
|
||||
}
|
||||
@@ -0,0 +1,34 @@
|
||||
use axum_test::TestServer;
|
||||
use ares_server::config::Config;
|
||||
use ares_server::state::AppState;
|
||||
use ares_server::store;
|
||||
use ares_server::router;
|
||||
|
||||
static TEST_DB_COUNTER: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
|
||||
|
||||
pub fn test_server() -> TestServer {
|
||||
let n = TEST_DB_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
||||
let db_path = std::env::temp_dir()
|
||||
.join(format!("ares_test_{}_{}.db", std::process::id(), n))
|
||||
.to_string_lossy()
|
||||
.into_owned();
|
||||
let cfg = Config { db_path, ..Default::default() };
|
||||
let conn = store::open(&cfg).unwrap();
|
||||
let state = AppState::new(cfg, conn);
|
||||
TestServer::new(router(state))
|
||||
}
|
||||
|
||||
pub async fn announce_test_file(server: &TestServer, name: &str, ftype: &str, hash: &str) {
|
||||
server.post("/api/v1/announce")
|
||||
.json(&serde_json::json!({
|
||||
"peer_id": "peer9",
|
||||
"name": name,
|
||||
"description": "",
|
||||
"type": ftype,
|
||||
"extension": name.rsplit('.').next().unwrap_or("bin"),
|
||||
"size_bytes": 100,
|
||||
"hash": hash,
|
||||
"license": "authored"
|
||||
}))
|
||||
.await;
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
use axum_test::TestServer;
|
||||
use ares_server::config::Config;
|
||||
use ares_server::{build_app, router};
|
||||
use tempfile::tempdir;
|
||||
|
||||
#[tokio::test]
|
||||
async fn health_returns_ok() {
|
||||
let dir = tempdir().unwrap();
|
||||
let cfg = Config {
|
||||
db_path: dir.path().join("t.db").to_string_lossy().into(),
|
||||
..Default::default()
|
||||
};
|
||||
let state = build_app(cfg).await;
|
||||
let server = TestServer::new(router(state));
|
||||
let res = server.get("/api/v1/health").await;
|
||||
assert_eq!(res.status_code(), 200);
|
||||
assert_eq!(res.json::<serde_json::Value>()["status"], "ok");
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
mod common;
|
||||
|
||||
use common::{announce_test_file, test_server};
|
||||
use serde_json::json;
|
||||
|
||||
#[tokio::test]
|
||||
async fn search_by_name_filters() {
|
||||
let server = test_server();
|
||||
for (name, ty, hash) in [
|
||||
("musica-bossa.ogg", "audio", "a1".repeat(32)),
|
||||
("musica-rock.mp3", "audio", "a2".repeat(32)),
|
||||
("ebook-p2p.pdf", "document", "a3".repeat(32)),
|
||||
] {
|
||||
announce_test_file(&server, name, ty, &hash).await;
|
||||
}
|
||||
let res = server.get("/api/v1/search?q=musica").await;
|
||||
assert_eq!(res.status_code(), 200);
|
||||
let items = res.json::<Vec<serde_json::Value>>();
|
||||
assert_eq!(items.len(), 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn search_filter_by_type() {
|
||||
let server = test_server();
|
||||
for (name, ty, hash) in [
|
||||
("musica-bossa.ogg", "audio", "a1".repeat(32)),
|
||||
("musica-rock.mp3", "audio", "a2".repeat(32)),
|
||||
("ebook-p2p.pdf", "document", "a3".repeat(32)),
|
||||
] {
|
||||
announce_test_file(&server, name, ty, &hash).await;
|
||||
}
|
||||
let res = server.get("/api/v1/search?q=musica&ftype=audio").await;
|
||||
let items = res.json::<Vec<serde_json::Value>>();
|
||||
assert_eq!(items.len(), 2);
|
||||
assert_eq!(items[0]["ftype"], "audio");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn search_pagination() {
|
||||
let server = test_server();
|
||||
for (name, ty, hash) in [
|
||||
("musica-bossa.ogg", "audio", "a1".repeat(32)),
|
||||
("musica-rock.mp3", "audio", "a2".repeat(32)),
|
||||
("ebook-p2p.pdf", "document", "a3".repeat(32)),
|
||||
] {
|
||||
announce_test_file(&server, name, ty, &hash).await;
|
||||
}
|
||||
let res = server.get("/api/v1/search?q=&limit=1&offset=0").await;
|
||||
let items = res.json::<Vec<serde_json::Value>>();
|
||||
assert_eq!(items.len(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn search_empty_results() {
|
||||
let server = test_server();
|
||||
let res = server.get("/api/v1/search?q=inexistente").await;
|
||||
assert_eq!(res.status_code(), 200);
|
||||
let items = res.json::<Vec<serde_json::Value>>();
|
||||
assert_eq!(items.len(), 0);
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
use ares_server::search::fts_query;
|
||||
|
||||
#[test]
|
||||
fn empty_query_returns_none() {
|
||||
assert_eq!(fts_query(""), None);
|
||||
assert_eq!(fts_query(" "), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn single_term_wrapped_in_quotes() {
|
||||
assert_eq!(fts_query("musica").unwrap(), "\"musica\"");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn multiple_terms_are_and() {
|
||||
assert_eq!(fts_query("bossa nova").unwrap(), "\"bossa\" AND \"nova\"");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn non_whitespace_tokens_split_kept() {
|
||||
// tokens are split by whitespace; edge chars (like *) are stripped
|
||||
assert_eq!(fts_query("*ideal* fluido*").unwrap(), "\"ideal\" AND \"fluido\"");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn fts_special_single_char_stripped_at_edges() {
|
||||
assert_eq!(fts_query("album' rock").unwrap(), "\"album\" AND \"rock\"");
|
||||
}
|
||||
@@ -0,0 +1,39 @@
|
||||
mod common;
|
||||
|
||||
use common::{announce_test_file, test_server};
|
||||
|
||||
#[tokio::test]
|
||||
async fn stats_reflects_files() {
|
||||
let server = test_server();
|
||||
announce_test_file(&server, "a.ogg", "audio", &"a1".repeat(32)).await;
|
||||
announce_test_file(&server, "b.ogg", "audio", &"a2".repeat(32)).await;
|
||||
announce_test_file(&server, "c.pdf", "document", &"a3".repeat(32)).await;
|
||||
let res = server.get("/api/v1/stats").await;
|
||||
assert_eq!(res.status_code(), 200);
|
||||
let body = res.json::<serde_json::Value>();
|
||||
assert_eq!(body["files"], 3);
|
||||
assert_eq!(body["total_bytes"], 300);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn announce_rate_limited() {
|
||||
let server = test_server();
|
||||
// 10 por segundo no RateLimiter padrão; dispara 12
|
||||
let mut last_status = axum::http::StatusCode::OK;
|
||||
for i in 0..12 {
|
||||
let res = server.post("/api/v1/announce")
|
||||
.json(&serde_json::json!({
|
||||
"peer_id": "ral",
|
||||
"name": format!("f{i}.ogg"),
|
||||
"description": "",
|
||||
"type": "audio",
|
||||
"extension": "ogg",
|
||||
"size_bytes": 10,
|
||||
"hash": format!("{:02x}", i).repeat(32),
|
||||
"license": "authored"
|
||||
}))
|
||||
.await;
|
||||
last_status = res.status_code();
|
||||
}
|
||||
assert_eq!(last_status, axum::http::StatusCode::BAD_REQUEST);
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
use ares_server::config::Config;
|
||||
use ares_server::store;
|
||||
use tempfile::tempdir;
|
||||
|
||||
#[test]
|
||||
fn migrate_creates_fts_table() {
|
||||
let dir = tempdir().unwrap();
|
||||
let cfg = Config {
|
||||
db_path: dir.path().join("t.db").to_string_lossy().into(),
|
||||
..Default::default()
|
||||
};
|
||||
let db = store::open(&cfg).expect("open db");
|
||||
let has_fts: i64 = db
|
||||
.query_row("SELECT count(*) FROM sqlite_master WHERE name='files_fts'", [], |r| r.get(0))
|
||||
.unwrap();
|
||||
assert_eq!(has_fts, 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn migrate_creates_files_table() {
|
||||
let dir = tempdir().unwrap();
|
||||
let cfg = Config {
|
||||
db_path: dir.path().join("t.db").to_string_lossy().into(),
|
||||
..Default::default()
|
||||
};
|
||||
let db = store::open(&cfg).expect("open db");
|
||||
let has_files: i64 = db
|
||||
.query_row("SELECT count(*) FROM sqlite_master WHERE name='files'", [], |r| r.get(0))
|
||||
.unwrap();
|
||||
assert_eq!(has_files, 1);
|
||||
}
|
||||
@@ -0,0 +1,72 @@
|
||||
use ares_server::config::Config;
|
||||
use ares_server::{build_app, router};
|
||||
use futures_util::{SinkExt, StreamExt};
|
||||
use serde_json::json;
|
||||
use tempfile::tempdir;
|
||||
|
||||
async fn spawn_server() -> (String, ares_server::state::SharedState) {
|
||||
let dir = tempdir().unwrap();
|
||||
let cfg = Config {
|
||||
db_path: dir.path().join("t.db").to_string_lossy().into(),
|
||||
..Default::default()
|
||||
};
|
||||
let state = build_app(cfg).await;
|
||||
let app = router(state.clone());
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
tokio::spawn(async move {
|
||||
let _ = axum::serve(listener, app).await;
|
||||
});
|
||||
(format!("http://{addr}"), state)
|
||||
}
|
||||
|
||||
fn ws_url(base: &str) -> String {
|
||||
base.replace("http", "ws")
|
||||
}
|
||||
|
||||
async fn connect_ws(base: &str, peer_id: &str, file_id: &str) -> tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>> {
|
||||
let (mut socket, _) =
|
||||
tokio_tungstenite::connect_async(&format!("{}/api/v1/tracker/ws", ws_url(base))).await.unwrap();
|
||||
socket
|
||||
.send(tokio_tungstenite::tungstenite::Message::Text(
|
||||
json!({"peer_id": peer_id, "file_id": file_id}).to_string(),
|
||||
)).await.unwrap();
|
||||
socket
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ws_registers_and_http_peers_lists_it() {
|
||||
let (base, _state) = spawn_server().await;
|
||||
let _socket = connect_ws(&base, "peerA", "abc123").await;
|
||||
|
||||
tokio::time::sleep(std::time::Duration::from_millis(200)).await;
|
||||
|
||||
let client = reqwest::Client::new();
|
||||
let res = client
|
||||
.get(format!("{base}/api/v1/peers/abc123"))
|
||||
.send().await.unwrap();
|
||||
let body: serde_json::Value = res.json().await.unwrap();
|
||||
assert_eq!(body["file_id"], "abc123");
|
||||
assert!(body["peers"].as_array().unwrap().contains(&serde_json::Value::String("peerA".into())));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ws_broadcasts_peer_online() {
|
||||
let (base, _state) = spawn_server().await;
|
||||
let mut a = connect_ws(&base, "peerA", "f1").await;
|
||||
tokio::time::sleep(std::time::Duration::from_millis(150)).await;
|
||||
|
||||
let _b = connect_ws(&base, "peerB", "f1").await;
|
||||
|
||||
// peerA deve receber a notificação de que peerB ficou online
|
||||
let timeout = tokio::time::timeout(std::time::Duration::from_secs(2), a.next());
|
||||
let got = timeout.await.unwrap().expect("peerA fechou").expect("erro ws");
|
||||
let text = match got {
|
||||
tokio_tungstenite::tungstenite::Message::Text(t) => t.to_string(),
|
||||
other => panic!("esperado Text, recebido: {other:?}"),
|
||||
};
|
||||
let parsed: serde_json::Value = serde_json::from_str(&text).unwrap();
|
||||
eprintln!("received: {parsed}");
|
||||
assert_eq!(parsed["type"], "peer_online");
|
||||
assert_eq!(parsed["peer_id"], "peerB");
|
||||
}
|
||||
Reference in New Issue
Block a user