feat: rewrite backend in Rust (axum)

Replace the Node/Express backend with a Rust implementation that keeps
the same REST API, MongoDB collections and Redis cache keys, so existing
users, subscriptions and progress keep working.

- axum + mongodb + redis, bcrypt and HS256 JWT compatible with old data
- RSS parsing, iTunes discovery/trending, background sync (1h / 4h)
- Docker image now builds a static Rust binary on debian-slim
- CI runs cargo tests before publishing the image
- compose uses pull_policy: always so a restart picks up the new image

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
MichaelandClaude Opus 5.5 committed 2026-10-13 09:34:33 +02:00
1 parent 0a91f6ec2c
commit 1028b7968c
41 files changed
+2280 -2239

No files matched your search

+24 -1
View File
@@ -2,14 +2,37 @@ name: Build & Publish Docker Image
on:
push:
branches: [main]
branches: [main, 'rust-*']
pull_request:
env:
REGISTRY: ghcr.io
IMAGE_NAME: ${{ github.repository }}
jobs:
test:
runs-on: ubuntu-latest
defaults:
run:
working-directory: backend
steps:
- name: Checkout repository
uses: actions/checkout@v4
- name: Install Rust toolchain
uses: dtolnay/rust-toolchain@stable
- name: Cache cargo
uses: Swatinem/rust-cache@v2
with:
workspaces: backend
- name: Run tests
run: cargo test
build-and-push:
needs: test
if: github.event_name == 'push' && github.ref == 'refs/heads/main'
runs-on: ubuntu-latest
permissions:
contents: read
+3
View File
@@ -40,3 +40,6 @@ coverage/
# Docker
.dockerignore
# Rust
target/
+21 -19
View File
@@ -8,41 +8,43 @@ RUN npm install
COPY frontend/ ./
# Build with /api as the API base URL (served by Express on same host)
# Build with /api as the API base URL (served by the Rust backend on same host)
ENV VITE_API_URL=/api
RUN npm run build
# Stage 2: Build backend
FROM node:20-alpine AS backend-builder
# Stage 2: Build Rust backend
FROM rust:1-bookworm AS backend-builder
WORKDIR /app
COPY backend/package*.json ./
RUN npm install
# Cache dependencies separately from sources
COPY backend/Cargo.toml backend/Cargo.lock* ./
RUN mkdir src && echo "fn main() {}" > src/main.rs \
&& cargo build --release \
&& rm -rf src
COPY backend/tsconfig.json ./
COPY backend/src ./src
RUN npm run build
RUN touch src/main.rs && cargo build --release
# Stage 3: Production image
FROM node:20-alpine
FROM debian:bookworm-slim
RUN apt-get update \
&& apt-get install -y --no-install-recommends ca-certificates \
&& rm -rf /var/lib/apt/lists/* \
&& useradd --system --no-create-home podcastic
WORKDIR /app
# Copy backend production deps
COPY backend/package*.json ./
RUN npm install --omit=dev
# Copy compiled backend
COPY --from=backend-builder /app/dist ./dist
# Copy frontend build into public/ (backend will serve it as static files)
COPY --from=backend-builder /app/target/release/podcastic ./podcastic
COPY --from=frontend-builder /frontend/dist ./public
USER podcastic
EXPOSE 3579
ENV NODE_ENV=production
ENV PORT=3579
ENV PUBLIC_DIR=/app/public
ENV RUST_LOG=podcastic=info,tower_http=warn
CMD ["node", "dist/index.js"]
CMD ["/app/podcastic"]
+4 -11
View File
@@ -63,8 +63,7 @@ docker-compose up -d
#### Backend
```bash
cd backend
npm install
npm run dev # Starts on port 5000
cargo run # Starts on port 5000 (set PORT=5000)
```
#### Frontend
@@ -87,15 +86,9 @@ Podcastic/
│ │ └── styles/
│ └── Dockerfile
│
├── backend/ # Express.js API
│ ├── src/
│ │ ├── routes/
│ │ ├── controllers/
│ │ ├── models/
│ │ ├── middleware/
│ │ ├── services/
│ │ └── config/
│ └── Dockerfile
├── backend/ # Rust (axum) API
│ ├── Cargo.toml
│ └── src/ # main, auth, podcasts, episodes, feeds, itunes, db
│
├── docker-compose.yml
├── .env.example
+4 -4
View File
@@ -1,12 +1,12 @@
# Podcastic - Unraid Installation Guide
Single container on port **3579** — includes backend, frontend, and serves everything via Express.
Single container on port **3579** — includes backend, frontend, and serves everything via the Rust backend.
## What's Included
This single Docker container includes:
- **Backend**: Node.js + Express.js API
- **Frontend**: React web interface (served by Express as static files)
- **Backend**: Rust (axum) API
- **Frontend**: React web interface (served by the Rust backend as static files)
- **Database**: MongoDB (separate container, managed by Compose)
- **Cache**: Redis (separate container, managed by Compose)
@@ -55,7 +55,7 @@ Open: `http://your-unraid-ip:3579`
```
docker-compose services:
├── podcastic (port 3579)
│ ├── Express.js API → /api/*
│ ├── Rust (axum) API → /api/*
│ └── React frontend → /* (static files)
├── podcastic-mongodb → internal :27017
└── podcastic-redis → internal :6379
+31
View File
@@ -0,0 +1,31 @@
[package]
name = "podcastic"
version = "2.0.0"
edition = "2021"
description = "Podcastic - Multi-user podcast streaming application backend"
license = "MIT"
[dependencies]
anyhow = "1.0.86"
axum = "0.7.5"
bcrypt = "0.15.1"
bson = { version = "2.11.0", features = ["chrono-0_4"] }
chrono = { version = "0.4.38", features = ["serde"] }
futures = "0.3.30"
jsonwebtoken = "9.3.0"
mongodb = "2.8.2"
redis = { version = "0.25.4", features = ["tokio-comp", "connection-manager"] }
reqwest = { version = "0.12.5", default-features = false, features = ["json", "rustls-tls", "gzip", "brotli", "deflate"] }
rss = "2.0.8"
serde = { version = "1.0.204", features = ["derive"] }
serde_json = "1.0.120"
tokio = { version = "1.38.0", features = ["full"] }
tower-http = { version = "0.5.2", features = ["fs", "cors", "set-header", "trace"] }
tracing = "0.1.40"
tracing-subscriber = { version = "0.3.18", features = ["env-filter"] }
[profile.release]
opt-level = 3
lto = "thin"
codegen-units = 1
strip = true
-16
View File
@@ -1,16 +0,0 @@
import js from '@eslint/js';
import tseslint from 'typescript-eslint';
export default tseslint.config(
js.configs.recommended,
...tseslint.configs.recommended,
{
rules: {
'@typescript-eslint/no-unused-vars': 'warn',
'@typescript-eslint/no-explicit-any': 'warn',
},
},
{
ignores: ['dist/**', 'node_modules/**'],
}
);
-45
View File
@@ -1,45 +0,0 @@
{
"name": "podcastic-backend",
"version": "1.0.0",
"description": "Podcastic - Multi-user podcast streaming application backend",
"main": "dist/index.js",
"scripts": {
"dev": "tsx watch src/index.ts",
"build": "tsc",
"start": "node dist/index.js",
"lint": "eslint src",
"test": "vitest"
},
"keywords": ["podcast", "streaming", "audio"],
"author": "",
"license": "MIT",
"dependencies": {
"express": "^4.18.2",
"mongoose": "^7.5.0",
"redis": "^4.6.11",
"jsonwebtoken": "^9.0.0",
"bcryptjs": "^2.4.3",
"dotenv": "^16.3.1",
"cors": "^2.8.5",
"zod": "^3.22.4",
"rss-parser": "^3.12.0",
"axios": "^1.5.0",
"node-cron": "^3.0.3",
"helmet": "^7.1.0",
"express-rate-limit": "^7.1.5"
},
"devDependencies": {
"@types/express": "^4.17.20",
"@types/node": "^20.8.0",
"@types/cors": "^2.8.17",
"@types/jsonwebtoken": "^9.0.4",
"@types/bcryptjs": "^2.4.2",
"@types/node-cron": "^3.0.11",
"typescript": "^5.2.2",
"tsx": "^4.1.0",
"eslint": "^9.18.0",
"@eslint/js": "^9.18.0",
"typescript-eslint": "^8.22.0",
"vitest": "^0.34.6"
}
}
+303
View File
@@ -0,0 +1,303 @@
use axum::{
async_trait,
extract::{FromRequestParts, State},
http::{header::AUTHORIZATION, request::Parts, StatusCode},
Json,
};
use bson::{doc, oid::ObjectId, Bson, Document};
use jsonwebtoken::{decode, encode, Algorithm, DecodingKey, EncodingKey, Header, Validation};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use crate::{
db::{self, coll, USERS},
error::{internal, issue, ApiResult, AppError},
AppState,
};
#[derive(Debug, Serialize, Deserialize)]
pub struct Claims {
pub id: String,
pub email: String,
pub iat: i64,
pub exp: i64,
}
pub struct JwtConfig {
encoding: EncodingKey,
decoding: DecodingKey,
access_ttl: i64,
refresh_ttl: i64,
}
impl JwtConfig {
pub fn new(secret: &str, access_ttl: &str, refresh_ttl: &str) -> Self {
Self {
encoding: EncodingKey::from_secret(secret.as_bytes()),
decoding: DecodingKey::from_secret(secret.as_bytes()),
access_ttl: parse_ttl(access_ttl).unwrap_or(7 * 86_400),
refresh_ttl: parse_ttl(refresh_ttl).unwrap_or(30 * 86_400),
}
}
fn sign(&self, id: &str, email: &str, ttl: i64) -> jsonwebtoken::errors::Result<String> {
let iat = chrono::Utc::now().timestamp();
let claims = Claims {
id: id.to_owned(),
email: email.to_owned(),
iat,
exp: iat + ttl,
};
encode(&Header::new(Algorithm::HS256), &claims, &self.encoding)
}
pub fn issue_pair(&self, id: &str, email: &str) -> jsonwebtoken::errors::Result<(String, String)> {
Ok((
self.sign(id, email, self.access_ttl)?,
self.sign(id, email, self.refresh_ttl)?,
))
}
pub fn verify(&self, token: &str) -> Option<Claims> {
decode::<Claims>(token, &self.decoding, &Validation::new(Algorithm::HS256))
.ok()
.map(|data| data.claims)
}
}
/// Parses `vercel/ms`-style durations used in env vars ("7d", "12h", "30m", "3600").
fn parse_ttl(value: &str) -> Option<i64> {
let value = value.trim();
let split = value
.find(|c: char| !c.is_ascii_digit())
.unwrap_or(value.len());
let (amount, unit) = value.split_at(split);
let amount: i64 = amount.parse().ok()?;
let factor = match unit.trim() {
"" | "s" => 1,
"m" => 60,
"h" => 3_600,
"d" => 86_400,
"w" => 604_800,
"y" => 31_557_600,
_ => return None,
};
Some(amount * factor)
}
/// Authenticated user extracted from the `Authorization: Bearer` header.
pub struct AuthUser {
pub id: ObjectId,
}
#[async_trait]
impl FromRequestParts<AppState> for AuthUser {
type Rejection = AppError;
async fn from_request_parts(parts: &mut Parts, state: &AppState) -> Result<Self, AppError> {
let token = parts
.headers
.get(AUTHORIZATION)
.and_then(|value| value.to_str().ok())
.and_then(|value| value.strip_prefix("Bearer "))
.ok_or_else(|| AppError::unauthorized("Missing or invalid authorization header"))?;
state
.jwt
.verify(token)
.and_then(|claims| db::oid(&claims.id))
.map(|id| AuthUser { id })
.ok_or_else(|| AppError::unauthorized("Invalid or expired token"))
}
}
/// Like [`AuthUser`] but never rejects the request.
pub struct MaybeUser(pub Option<AuthUser>);
#[async_trait]
impl FromRequestParts<AppState> for MaybeUser {
type Rejection = AppError;
async fn from_request_parts(parts: &mut Parts, state: &AppState) -> Result<Self, AppError> {
Ok(MaybeUser(AuthUser::from_request_parts(parts, state).await.ok()))
}
}
fn str_field(body: &Value, key: &str) -> String {
body.get(key)
.and_then(Value::as_str)
.unwrap_or_default()
.to_owned()
}
fn is_valid_email(email: &str) -> bool {
let Some((local, domain)) = email.split_once('@') else {
return false;
};
!local.is_empty()
&& !domain.contains('@')
&& domain.contains('.')
&& !domain.starts_with('.')
&& !domain.ends_with('.')
&& !email.chars().any(char::is_whitespace)
}
fn user_json(mut user: Document) -> Value {
user.remove("password");
db::doc_to_json(user)
}
fn auth_response(state: &AppState, user: Document, message: &str) -> ApiResult<Value> {
let id = user
.get_object_id("_id")
.map_err(internal("Authentication failed"))?
.to_hex();
let email = user.get_str("email").unwrap_or_default().to_owned();
let (access_token, refresh_token) = state
.jwt
.issue_pair(&id, &email)
.map_err(internal("Authentication failed"))?;
Ok(json!({
"message": message,
"user": user_json(user),
"tokens": { "accessToken": access_token, "refreshToken": refresh_token },
}))
}
pub async fn register(
State(state): State<AppState>,
Json(body): Json<Value>,
) -> ApiResult<(StatusCode, Json<Value>)> {
let email = str_field(&body, "email");
let password = str_field(&body, "password");
let username = str_field(&body, "username");
let mut errors = Vec::new();
if !is_valid_email(email.trim()) {
errors.push(issue("email", "Invalid email address"));
}
if password.chars().count() < 6 {
errors.push(issue("password", "Password must be at least 6 characters"));
}
if username.chars().count() < 2 {
errors.push(issue("username", "Username must be at least 2 characters"));
}
if !errors.is_empty() {
return Err(AppError::validation(errors));
}
let email = email.trim().to_lowercase();
let users = coll(&state.db, USERS);
let existing = users
.find_one(doc! { "email": email.as_str() }, None)
.await
.map_err(internal("Registration failed"))?;
if existing.is_some() {
return Err(AppError::new(StatusCode::CONFLICT, "Email already registered"));
}
let hash = tokio::task::spawn_blocking(move || bcrypt::hash(password, 10))
.await
.map_err(internal("Registration failed"))?
.map_err(internal("Registration failed"))?;
let now = db::now();
let mut user = doc! {
"email": email.as_str(),
"password": hash,
"username": username.trim(),
"avatar": Bson::Null,
"preferences": { "defaultPlaybackSpeed": 1, "theme": "light", "language": "en" },
"createdAt": now,
"updatedAt": now,
"__v": 0,
};
let inserted = users.insert_one(&user, None).await.map_err(|err| {
if db::is_duplicate_key(&err) {
AppError::new(StatusCode::CONFLICT, "Email already registered")
} else {
internal("Registration failed")(err)
}
})?;
user.insert("_id", inserted.inserted_id);
let body = auth_response(&state, user, "User registered successfully")?;
Ok((StatusCode::CREATED, Json(body)))
}
pub async fn login(State(state): State<AppState>, Json(body): Json<Value>) -> ApiResult {
let email = str_field(&body, "email");
let password = str_field(&body, "password");
let mut errors = Vec::new();
if !is_valid_email(email.trim()) {
errors.push(issue("email", "Invalid email address"));
}
if password.is_empty() {
errors.push(issue("password", "Password required"));
}
if !errors.is_empty() {
return Err(AppError::validation(errors));
}
let invalid = || AppError::unauthorized("Invalid email or password");
let user = coll(&state.db, USERS)
.find_one(doc! { "email": email.trim().to_lowercase() }, None)
.await
.map_err(internal("Login failed"))?
.ok_or_else(invalid)?;
let hash = user.get_str("password").unwrap_or_default().to_owned();
let valid = tokio::task::spawn_blocking(move || bcrypt::verify(password, &hash).unwrap_or(false))
.await
.map_err(internal("Login failed"))?;
if !valid {
return Err(invalid());
}
Ok(Json(auth_response(&state, user, "Login successful")?))
}
pub async fn me(State(state): State<AppState>, user: AuthUser) -> ApiResult {
let user = coll(&state.db, USERS)
.find_one(doc! { "_id": user.id }, None)
.await
.map_err(internal("Failed to fetch user"))?
.ok_or_else(|| AppError::not_found("User not found"))?;
Ok(Json(json!({ "user": user_json(user) })))
}
pub async fn logout(_user: AuthUser) -> Json<Value> {
// Stateless JWT: the client simply drops its tokens.
Json(json!({ "message": "Logout successful" }))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn parses_ttl() {
assert_eq!(parse_ttl("7d"), Some(604_800));
assert_eq!(parse_ttl("15m"), Some(900));
assert_eq!(parse_ttl("3600"), Some(3600));
assert_eq!(parse_ttl("abc"), None);
}
#[test]
fn round_trips_tokens() {
let jwt = JwtConfig::new("secret", "7d", "30d");
let (access, _) = jwt.issue_pair("65f000000000000000000000", "a@b.co").unwrap();
let claims = jwt.verify(&access).unwrap();
assert_eq!(claims.email, "a@b.co");
assert!(jwt.verify("garbage").is_none());
}
#[test]
fn validates_email() {
assert!(is_valid_email("john@example.com"));
assert!(!is_valid_email("john@example"));
assert!(!is_valid_email("john.example.com"));
}
}
-28
View File
@@ -1,28 +0,0 @@
import mongoose from 'mongoose';
export const connectDB = async () => {
try {
const mongoUri = process.env.MONGODB_URI || 'mongodb://localhost:27017/podcastic';
await mongoose.connect(mongoUri, {
retryWrites: true,
w: 'majority',
});
console.log('✓ MongoDB connected successfully');
return mongoose.connection;
} catch (error) {
console.error('✗ MongoDB connection failed:', error);
process.exit(1);
}
};
export const disconnectDB = async () => {
try {
await mongoose.disconnect();
console.log('✓ MongoDB disconnected');
} catch (error) {
console.error('✗ MongoDB disconnection failed:', error);
process.exit(1);
}
};
-46
View File
@@ -1,46 +0,0 @@
import { createClient } from 'redis';
let redisClient: ReturnType<typeof createClient> | null = null;
export const initRedis = async () => {
try {
const redisUrl = process.env.REDIS_URL || 'redis://localhost:6379';
redisClient = createClient({
url: redisUrl,
socket: {
reconnectStrategy: (retries) => {
if (retries > 10) {
console.error('✗ Redis max retries reached');
return new Error('Max retries');
}
return retries * 50;
},
},
});
redisClient.on('error', (err) => console.error('Redis error:', err));
redisClient.on('connect', () => console.log('✓ Redis connected'));
redisClient.on('reconnecting', () => console.log('→ Redis reconnecting...'));
await redisClient.connect();
return redisClient;
} catch (error) {
console.error('✗ Redis connection failed:', error);
process.exit(1);
}
};
export const getRedis = () => {
if (!redisClient) {
throw new Error('Redis not initialized');
}
return redisClient;
};
export const closeRedis = async () => {
if (redisClient) {
await redisClient.quit();
console.log('✓ Redis connection closed');
}
};
-121
View File
@@ -1,121 +0,0 @@
import { Request, Response } from 'express';
import { User } from '../models/User';
import { generateTokens } from '../utils/jwt';
import { z } from 'zod';
const registerSchema = z.object({
email: z.string().email('Invalid email address'),
password: z.string().min(6, 'Password must be at least 6 characters'),
username: z.string().min(2, 'Username must be at least 2 characters'),
});
const loginSchema = z.object({
email: z.string().email('Invalid email address'),
password: z.string().min(1, 'Password required'),
});
export const register = async (req: Request, res: Response) => {
try {
const { email, password, username } = registerSchema.parse(req.body);
// Check if user already exists
const existingUser = await User.findOne({ email });
if (existingUser) {
return res.status(409).json({ message: 'Email already registered' });
}
// Create new user
const user = new User({
email,
password,
username,
});
await user.save();
// Generate tokens
const { accessToken, refreshToken } = generateTokens(user);
res.status(201).json({
message: 'User registered successfully',
user: user.toJSON(),
tokens: { accessToken, refreshToken },
});
} catch (error) {
if (error instanceof z.ZodError) {
return res.status(400).json({
message: 'Validation error',
errors: error.errors,
});
}
console.error('Register error:', error);
res.status(500).json({ message: 'Registration failed' });
}
};
export const login = async (req: Request, res: Response) => {
try {
const { email, password } = loginSchema.parse(req.body);
// Find user
const user = await User.findOne({ email });
if (!user) {
return res.status(401).json({ message: 'Invalid email or password' });
}
// Verify password
const isPasswordValid = await user.comparePassword(password);
if (!isPasswordValid) {
return res.status(401).json({ message: 'Invalid email or password' });
}
// Generate tokens
const { accessToken, refreshToken } = generateTokens(user);
res.json({
message: 'Login successful',
user: user.toJSON(),
tokens: { accessToken, refreshToken },
});
} catch (error) {
if (error instanceof z.ZodError) {
return res.status(400).json({
message: 'Validation error',
errors: error.errors,
});
}
console.error('Login error:', error);
res.status(500).json({ message: 'Login failed' });
}
};
export const getMe = async (req: Request, res: Response) => {
try {
if (!req.user) {
return res.status(401).json({ message: 'Not authenticated' });
}
const user = await User.findById(req.user.id);
if (!user) {
return res.status(404).json({ message: 'User not found' });
}
res.json({ user: user.toJSON() });
} catch (error) {
console.error('GetMe error:', error);
res.status(500).json({ message: 'Failed to fetch user' });
}
};
export const logout = async (req: Request, res: Response) => {
try {
// In a stateless JWT system, logout is client-side (delete token)
// Could implement token blacklist with Redis if needed
res.json({ message: 'Logout successful' });
} catch (error) {
console.error('Logout error:', error);
res.status(500).json({ message: 'Logout failed' });
}
};
@@ -1,224 +0,0 @@
import { Request, Response } from 'express';
import { Episode } from '../models/Episode';
import { UserProgress } from '../models/UserProgress';
import { UserSubscription } from '../models/UserSubscription';
export const getLatestEpisodes = async (req: Request, res: Response) => {
try {
if (!req.user) {
return res.status(401).json({ message: 'Not authenticated' });
}
const { limit = 30, skip = 0 } = req.query;
// Get user's subscribed podcasts
const subscriptions = await UserSubscription.find({
userId: req.user.id,
}).select('podcastId');
const podcastIds = subscriptions.map((sub) => sub.podcastId);
if (podcastIds.length === 0) {
return res.json({ episodes: [], count: 0 });
}
// Get latest episodes from subscribed podcasts
const episodes = await Episode.find({
podcastId: { $in: podcastIds },
})
.populate('podcastId', 'title author imageUrl')
.sort({ pubDate: -1 })
.limit(Number(limit))
.skip(Number(skip));
// Fetch progress for these episodes
const episodeIds = episodes.map((e) => e._id);
const progresses = await UserProgress.find({
userId: req.user.id,
episodeId: { $in: episodeIds },
});
const episodesWithProgress = episodes.map((episode) => {
const progress = progresses.find(
(p) => p.episodeId.toString() === episode._id.toString()
);
return {
...episode.toJSON(),
progress: progress ? { position: progress.position, isCompleted: progress.isCompleted } : null,
};
});
res.json({ episodes: episodesWithProgress, count: episodesWithProgress.length });
} catch (error) {
console.error('Get latest episodes error:', error);
res.status(500).json({ message: 'Failed to fetch episodes' });
}
};
export const getPodcastEpisodes = async (req: Request, res: Response) => {
try {
const { podcastId } = req.params;
const { limit = 50, skip = 0 } = req.query;
const episodes = await Episode.find({ podcastId })
.sort({ pubDate: -1 })
.limit(Number(limit))
.skip(Number(skip));
// If user is authenticated, fetch progress
let episodesWithProgress = episodes.map(e => e.toJSON());
if (req.user) {
const progresses = await UserProgress.find({
userId: req.user.id,
episodeId: { $in: episodes.map(e => e._id) }
});
episodesWithProgress = episodes.map((episode) => {
const progress = progresses.find(
(p) => p.episodeId.toString() === episode._id.toString()
);
return {
...episode.toJSON(),
progress: progress ? { position: progress.position, isCompleted: progress.isCompleted } : null,
};
});
}
res.json({ episodes: episodesWithProgress, count: episodes.length });
} catch (error) {
console.error('Get podcast episodes error:', error);
res.status(500).json({ message: 'Failed to fetch episodes' });
}
};
export const getEpisode = async (req: Request, res: Response) => {
try {
const { episodeId } = req.params;
const episode = await Episode.findById(episodeId).populate(
'podcastId',
'title author description imageUrl'
);
if (!episode) {
return res.status(404).json({ message: 'Episode not found' });
}
res.json({ episode });
} catch (error) {
console.error('Get episode error:', error);
res.status(500).json({ message: 'Failed to fetch episode' });
}
};
export const saveProgress = async (req: Request, res: Response) => {
try {
if (!req.user) {
return res.status(401).json({ message: 'Not authenticated' });
}
const { episodeId, position, isCompleted } = req.body;
if (!episodeId || position === undefined) {
return res.status(400).json({ message: 'Episode ID and position required' });
}
// Find or create progress record
let progress = await UserProgress.findOne({
userId: req.user.id,
episodeId,
});
if (progress) {
progress.position = position;
if (isCompleted !== undefined) {
progress.isCompleted = isCompleted;
}
progress.lastListenedAt = new Date();
} else {
progress = new UserProgress({
userId: req.user.id,
episodeId,
position,
isCompleted: isCompleted || false,
lastListenedAt: new Date(),
});
}
await progress.save();
res.json({
message: 'Progress saved',
progress,
});
} catch (error) {
console.error('Save progress error:', error);
res.status(500).json({ message: 'Failed to save progress' });
}
};
export const getProgress = async (req: Request, res: Response) => {
try {
if (!req.user) {
return res.status(401).json({ message: 'Not authenticated' });
}
const { episodeId } = req.params;
const progress = await UserProgress.findOne({
userId: req.user.id,
episodeId,
});
if (!progress) {
return res.json({ progress: null, message: 'No progress found' });
}
res.json({ progress });
} catch (error) {
console.error('Get progress error:', error);
res.status(500).json({ message: 'Failed to fetch progress' });
}
};
export const searchEpisodes = async (req: Request, res: Response) => {
try {
const { q, limit = 40, skip = 0 } = req.query;
if (!q) {
return res.status(400).json({ message: 'Search query required' });
}
// Search episodes by title/description using text index
const episodes = await Episode.find(
{ $text: { $search: q as string } },
{ score: { $meta: 'textScore' } }
)
.populate('podcastId', 'title author imageUrl')
.sort({ score: { $meta: 'textScore' } })
.limit(Number(limit))
.skip(Number(skip));
// Fetch progress if authenticated
let episodesWithProgress = episodes.map(e => e.toJSON());
if (req.user) {
const progresses = await UserProgress.find({
userId: req.user.id,
episodeId: { $in: episodes.map(e => e._id) }
});
episodesWithProgress = episodes.map((episode) => {
const progress = progresses.find(
(p) => p.episodeId.toString() === episode._id.toString()
);
return {
...episode.toJSON(),
progress: progress ? { position: progress.position, isCompleted: progress.isCompleted } : null,
};
});
}
res.json({ episodes: episodesWithProgress, count: episodesWithProgress.length });
} catch (error) {
console.error('Search episodes error:', error);
res.status(500).json({ message: 'Search failed' });
}
};
@@ -1,328 +0,0 @@
import { Request, Response } from 'express';
import mongoose from 'mongoose';
import { Podcast } from '../models/Podcast';
import { UserSubscription } from '../models/UserSubscription';
import { Episode } from '../models/Episode';
import { podcastIndexService } from '../services/podcastIndexService';
import { rssParserService } from '../services/rssParserService';
import { z } from 'zod';
const subscribeSchema = z.object({
rssUrl: z.string().url('Invalid RSS URL'),
});
const searchSchema = z.object({
q: z.string().min(2, 'Query must be at least 2 characters'),
limit: z.coerce.number().int().min(1).max(50).optional().default(20),
offset: z.coerce.number().int().min(0).optional().default(0),
});
export const getUserSubscriptions = async (req: Request, res: Response) => {
try {
if (!req.user) {
return res.status(401).json({ message: 'Not authenticated' });
}
const subscriptions = await UserSubscription.find({
userId: req.user.id,
})
.populate('podcastId')
.sort({ subscribedAt: -1 });
const podcasts = subscriptions.map((sub) => {
const podcastObj = (sub.podcastId as any).toJSON();
return {
...podcastObj,
subscriptionId: sub._id.toString(),
};
});
res.json({ podcasts, count: podcasts.length });
} catch (error) {
console.error('Get subscriptions error:', error);
res.status(500).json({ message: 'Failed to fetch subscriptions' });
}
};
export const subscribe = async (req: Request, res: Response) => {
try {
if (!req.user) {
return res.status(401).json({ message: 'Not authenticated' });
}
const { rssUrl } = subscribeSchema.parse(req.body);
// Check if podcast exists
let podcast = await Podcast.findOne({ rssUrl });
if (!podcast) {
// Parse RSS feed
const feedData = await rssParserService.parseFeed(rssUrl);
const episodes = rssParserService.parseEpisodes(feedData.episodes, feedData.imageUrl, rssUrl);
try {
const result = await rssParserService.createPodcastFromRss(rssUrl);
podcast = await Podcast.findById(result.podcast._id);
} catch (error: any) {
if (error.message.includes('already in database')) {
podcast = await Podcast.findOne({ rssUrl });
} else {
return res.status(400).json({
message: `Impossible de récupérer le podcast : ${error.message}`,
});
}
}
}
if (!podcast) {
return res.status(400).json({ message: "Impossible d'ajouter le podcast" });
}
// Check if already subscribed
const existingSubscription = await UserSubscription.findOne({
userId: req.user.id,
podcastId: podcast._id,
});
if (!existingSubscription) {
const subscription = new UserSubscription({
userId: req.user.id,
podcastId: podcast._id,
});
await subscription.save();
}
res.status(200).json({
message: existingSubscription ? 'Déjà abonné' : 'Abonnement réussi',
podcast: podcast.toJSON(),
});
} catch (error) {
if (error instanceof z.ZodError) {
return res.status(400).json({
message: 'Validation error',
errors: error.errors,
});
}
console.error('Subscribe error:', error);
res.status(500).json({ message: "Impossible de s'abonner" });
}
};
export const unsubscribe = async (req: Request, res: Response) => {
try {
if (!req.user) {
return res.status(401).json({ message: 'Not authenticated' });
}
const { podcastId } = req.params;
// Log the request
console.log(`[Unsubscribe] User ${req.user.id} attempting to remove: ${podcastId}`);
// Method 1: Try deletion by Subscription ID (New robust method)
let subscription = await UserSubscription.findOneAndDelete({
_id: new mongoose.Types.ObjectId(podcastId),
userId: new mongoose.Types.ObjectId(req.user.id),
});
// Method 2: Fallback to Podcast ID if Method 1 found nothing (Old compatibility method)
if (!subscription) {
console.log(`[Unsubscribe] Not found by Subscription ID, trying by Podcast ID: ${podcastId}`);
subscription = await UserSubscription.findOneAndDelete({
userId: new mongoose.Types.ObjectId(req.user.id),
podcastId: new mongoose.Types.ObjectId(podcastId),
});
}
if (!subscription) {
console.warn(`[Unsubscribe] No subscription record found for deletion.`);
return res.status(404).json({ message: 'Subscription not found' });
}
// Cast to any to avoid TypeScript error with ModifyResult in logs
const result = subscription as any;
console.log(`[Unsubscribe] Successfully removed subscription ${result._id}`);
res.json({ message: 'Unsubscribed successfully', deletedId: result._id });
} catch (error) {
console.error('Unsubscribe error:', error);
res.status(500).json({ message: 'Failed to unsubscribe' });
}
};
export const syncPodcast = async (req: Request, res: Response) => {
try {
if (!req.user) {
return res.status(401).json({ message: 'Not authenticated' });
}
const { podcastId } = req.params;
const podcast = await Podcast.findById(podcastId);
if (!podcast) {
return res.status(404).json({ message: 'Podcast not found' });
}
console.log(`[Sync] Triggering manual sync for podcast: ${podcast.title} (${podcastId})`);
const result = await rssParserService.syncPodcastEpisodes(podcastId, podcast.rssUrl, true);
res.json(result);
} catch (error: any) {
console.error('Manual sync error:', error);
res.status(500).json({ message: error.message || 'Failed to sync podcast' });
}
};
export const searchPodcasts = async (req: Request, res: Response) => {
try {
const { q, limit = 20, skip = 0 } = req.query;
if (!q) {
return res.status(400).json({ message: 'Search query required' });
}
const podcasts = await Podcast.find(
{ $text: { $search: q as string } },
{ score: { $meta: 'textScore' } }
)
.sort({ score: { $meta: 'textScore' } })
.limit(Number(limit))
.skip(Number(skip));
res.json({ podcasts, count: podcasts.length });
} catch (error) {
console.error('Search error:', error);
res.status(500).json({ message: 'Search failed' });
}
};
export const getPodcast = async (req: Request, res: Response) => {
try {
const { podcastId } = req.params;
const podcast = await Podcast.findById(podcastId);
if (!podcast) {
return res.status(404).json({ message: 'Podcast not found' });
}
res.json({ podcast });
} catch (error) {
console.error('Get podcast error:', error);
res.status(500).json({ message: 'Failed to fetch podcast' });
}
};
export const discoverPodcasts = async (req: Request, res: Response) => {
try {
const { q, limit = 20 } = searchSchema.parse(req.query);
const query = q as string;
const lim = limit as number;
// 1. Try iTunes Search API
try {
const results = await podcastIndexService.searchPodcasts(query, lim);
if (results.length > 0) {
return res.json({ source: 'itunes', podcasts: results, count: results.length });
}
console.log(`[Discover] iTunes returned 0 results for "${query}", trying local DB`);
} catch (itunesError: any) {
console.error(`[Discover] iTunes failed for "${query}":`, itunesError.message);
}
// 2. Fallback: search local DB podcasts already in the system
const localPodcasts = await Podcast.find({
$or: [
{ title: { $regex: query, $options: 'i' } },
{ author: { $regex: query, $options: 'i' } },
{ description: { $regex: query, $options: 'i' } },
],
}).limit(lim);
const mapped = localPodcasts.map((p: any) => ({
id: p._id.toString(),
title: p.title,
rssUrl: p.rssUrl,
description: p.description || '',
author: p.author || '',
imageUrl: p.imageUrl,
episodeCount: p.episodeCount,
}));
return res.json({ source: 'local', podcasts: mapped, count: mapped.length });
} catch (error: any) {
console.error('Discover podcasts error:', error.message);
res.json({ source: 'itunes', podcasts: [], count: 0 });
}
};
export const subscribeFromDiscovery = async (req: Request, res: Response) => {
try {
if (!req.user) {
return res.status(401).json({ message: 'Not authenticated' });
}
const { rssUrl, title, author, imageUrl } = req.body;
if (!rssUrl) {
return res.status(400).json({ message: 'RSS URL required' });
}
// Check if podcast already in database
let podcast = await Podcast.findOne({ rssUrl });
// If not in database, create it from RSS (parse feed for real episodes)
if (!podcast) {
try {
const result = await rssParserService.createPodcastFromRss(rssUrl);
// Refetch as a proper Mongoose document (result.podcast is a plain object)
podcast = await Podcast.findById(result.podcast._id);
} catch (error: any) {
if (error.message.includes('already in database')) {
podcast = await Podcast.findOne({ rssUrl });
} else {
return res.status(400).json({
message: `Impossible de récupérer le podcast : ${error.message}`,
});
}
}
}
if (!podcast) {
return res.status(400).json({ message: "Impossible d'ajouter le podcast" });
}
// Check if already subscribed
const existingSubscription = await UserSubscription.findOne({
userId: req.user.id,
podcastId: podcast._id,
});
if (!existingSubscription) {
const subscription = new UserSubscription({
userId: req.user.id,
podcastId: podcast._id,
});
await subscription.save();
}
res.status(200).json({
message: existingSubscription ? 'Déjà abonné' : 'Abonnement réussi',
podcast: podcast.toJSON(),
});
} catch (error: any) {
console.error('Subscribe from discovery error:', error);
res.status(500).json({ message: "Impossible de s'abonner" });
}
};
export const getTrendingPodcasts = async (req: Request, res: Response) => {
try {
const limit = Number(req.query.limit) || 20;
const genre = req.query.genre as string;
const trending = await podcastIndexService.getTrendingPodcasts(limit, genre);
res.json({ source: 'itunes', podcasts: trending, count: trending.length });
} catch (error: any) {
console.error('Trending podcasts error:', error.message);
// Return empty rather than crashing the page
res.json({ source: 'itunes', podcasts: [], count: 0 });
}
};
+172
View File
@@ -0,0 +1,172 @@
//! MongoDB access. Collection names and document shapes match the former
//! Mongoose models so existing data keeps working after the Rust rewrite.
use std::{collections::HashMap, time::Duration};
use bson::{doc, oid::ObjectId, Bson, Document};
use futures::TryStreamExt;
use mongodb::{
error::{Error as MongoError, ErrorKind, WriteFailure},
options::{ClientOptions, FindOptions, IndexOptions},
Client, Collection, Database, IndexModel,
};
use serde_json::{Map, Value};
pub const USERS: &str = "users";
pub const PODCASTS: &str = "podcasts";
pub const EPISODES: &str = "episodes";
pub const PROGRESS: &str = "userprogresses";
pub const SUBSCRIPTIONS: &str = "usersubscriptions";
pub async fn connect(uri: &str) -> anyhow::Result<Database> {
let mut options = ClientOptions::parse(uri).await?;
options.app_name = Some("podcastic".into());
let client = Client::with_options(options)?;
let db = client
.default_database()
.unwrap_or_else(|| client.database("podcastic"));
let mut attempt = 1;
loop {
match db.run_command(doc! { "ping": 1 }, None).await {
Ok(_) => break,
Err(err) if attempt < 15 => {
tracing::warn!("MongoDB not ready (attempt {attempt}): {err}");
attempt += 1;
tokio::time::sleep(Duration::from_secs(2)).await;
}
Err(err) => return Err(err.into()),
}
}
ensure_indexes(&db).await;
tracing::info!("✓ MongoDB connected successfully");
Ok(db)
}
async fn ensure_indexes(db: &Database) {
let specs: Vec<(&str, Document, bool)> = vec![
(USERS, doc! { "email": 1 }, true),
(PODCASTS, doc! { "rssUrl": 1 }, true),
(PODCASTS, doc! { "title": "text", "description": "text", "author": "text" }, false),
(PODCASTS, doc! { "category": 1 }, false),
(EPISODES, doc! { "podcastId": 1, "guid": 1 }, true),
(EPISODES, doc! { "podcastId": 1 }, false),
(EPISODES, doc! { "pubDate": -1 }, false),
(EPISODES, doc! { "title": "text", "description": "text" }, false),
(PROGRESS, doc! { "userId": 1, "episodeId": 1 }, true),
(PROGRESS, doc! { "userId": 1, "lastListenedAt": -1 }, false),
(SUBSCRIPTIONS, doc! { "userId": 1, "podcastId": 1 }, true),
(SUBSCRIPTIONS, doc! { "userId": 1, "isFavorite": 1 }, false),
];
for (name, keys, unique) in specs {
let options = unique.then(|| IndexOptions::builder().unique(true).build());
let model = IndexModel::builder().keys(keys).options(options).build();
// Indexes created earlier by Mongoose may differ slightly; that is fine.
if let Err(err) = coll(db, name).create_index(model, None).await {
tracing::debug!("Index on {name} skipped: {err}");
}
}
}
pub fn coll(db: &Database, name: &str) -> Collection<Document> {
db.collection::<Document>(name)
}
pub fn oid(value: &str) -> Option<ObjectId> {
ObjectId::parse_str(value.trim()).ok()
}
pub fn now() -> bson::DateTime {
bson::DateTime::now()
}
pub fn iso(dt: bson::DateTime) -> String {
dt.to_chrono()
.to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
}
pub async fn find_all(
collection: &Collection<Document>,
filter: Document,
options: impl Into<Option<FindOptions>>,
) -> mongodb::error::Result<Vec<Document>> {
collection.find(filter, options).await?.try_collect().await
}
pub fn is_duplicate_key(err: &MongoError) -> bool {
matches!(
err.kind.as_ref(),
ErrorKind::Write(WriteFailure::WriteError(write_error)) if write_error.code == 11000
)
}
/// Converts BSON to the JSON shape Mongoose used to emit
/// (ObjectIds as hex strings, dates as ISO-8601 strings).
pub fn to_json(value: Bson) -> Value {
match value {
Bson::ObjectId(id) => Value::String(id.to_hex()),
Bson::DateTime(dt) => Value::String(iso(dt)),
Bson::Document(document) => doc_to_json(document),
Bson::Array(items) => Value::Array(items.into_iter().map(to_json).collect()),
Bson::String(s) => Value::String(s),
Bson::Boolean(b) => Value::Bool(b),
Bson::Int32(i) => Value::from(i),
Bson::Int64(i) => Value::from(i),
Bson::Double(f) if f.fract() == 0.0 && f.abs() < 9e15 => Value::from(f as i64),
Bson::Double(f) => serde_json::Number::from_f64(f)
.map(Value::Number)
.unwrap_or(Value::Null),
Bson::Null | Bson::Undefined => Value::Null,
other => other.into_relaxed_extjson(),
}
}
pub fn doc_to_json(document: Document) -> Value {
Value::Object(
document
.into_iter()
.map(|(key, value)| (key, to_json(value)))
.collect::<Map<String, Value>>(),
)
}
/// Mongoose-style `populate('podcastId', fields)`: replaces each episode's
/// `podcastId` with the selected podcast fields, or null when missing.
pub async fn populate_podcasts(
db: &Database,
episodes: &mut [Document],
fields: &[&str],
) -> mongodb::error::Result<()> {
let ids: Vec<ObjectId> = episodes
.iter()
.filter_map(|e| e.get_object_id("podcastId").ok())
.collect();
if ids.is_empty() {
return Ok(());
}
let mut projection = Document::new();
for field in fields {
projection.insert(*field, 1);
}
let options = FindOptions::builder().projection(projection).build();
let podcasts: HashMap<ObjectId, Document> =
find_all(&coll(db, PODCASTS), doc! { "_id": { "$in": ids } }, options)
.await?
.into_iter()
.filter_map(|p| Some((p.get_object_id("_id").ok()?, p)))
.collect();
for episode in episodes.iter_mut() {
let populated = episode
.get_object_id("podcastId")
.ok()
.and_then(|id| podcasts.get(&id).cloned())
.map(Bson::Document)
.unwrap_or(Bson::Null);
episode.insert("podcastId", populated);
}
Ok(())
}
+237
View File
@@ -0,0 +1,237 @@
use std::collections::HashMap;
use axum::{
extract::{Path, Query, State},
Json,
};
use bson::{doc, oid::ObjectId, Bson, Document};
use mongodb::options::{FindOneAndUpdateOptions, FindOptions, ReturnDocument};
use serde_json::{json, Value};
use crate::{
auth::{AuthUser, MaybeUser},
db::{self, coll, find_all, populate_podcasts, EPISODES, PROGRESS, SUBSCRIPTIONS},
error::{internal, ApiResult, AppError},
podcasts::num_param,
AppState,
};
type Params = Query<HashMap<String, String>>;
const LIST_FIELDS: &[&str] = &["title", "author", "imageUrl"];
/// Serializes episodes, adding `progress: { position, isCompleted } | null` for `user`.
async fn with_progress(
state: &AppState,
user: Option<&AuthUser>,
episodes: Vec<Document>,
) -> mongodb::error::Result<Vec<Value>> {
let Some(user) = user else {
return Ok(episodes.into_iter().map(db::doc_to_json).collect());
};
let ids: Vec<ObjectId> = episodes
.iter()
.filter_map(|e| e.get_object_id("_id").ok())
.collect();
let progress: HashMap<ObjectId, Value> = find_all(
&coll(&state.db, PROGRESS),
doc! { "userId": user.id, "episodeId": { "$in": ids } },
None,
)
.await?
.into_iter()
.filter_map(|p| {
let id = p.get_object_id("episodeId").ok()?;
let position = p.get("position").cloned().map(db::to_json).unwrap_or(json!(0));
let completed = p.get_bool("isCompleted").unwrap_or(false);
Some((id, json!({ "position": position, "isCompleted": completed })))
})
.collect();
Ok(episodes
.into_iter()
.map(|episode| {
let entry = episode
.get_object_id("_id")
.ok()
.and_then(|id| progress.get(&id).cloned())
.unwrap_or(Value::Null);
let mut value = db::doc_to_json(episode);
value["progress"] = entry;
value
})
.collect())
}
fn list_options(params: &HashMap<String, String>, default_limit: i64, sort: Document) -> FindOptions {
FindOptions::builder()
.sort(sort)
.skip(num_param(params, "skip", 0) as u64)
.limit(num_param(params, "limit", default_limit))
.build()
}
pub async fn latest(
State(state): State<AppState>,
user: AuthUser,
Query(params): Params,
) -> ApiResult {
const ERR: &str = "Failed to fetch episodes";
let podcast_ids: Vec<Bson> = find_all(&coll(&state.db, SUBSCRIPTIONS), doc! { "userId": user.id }, None)
.await
.map_err(internal(ERR))?
.into_iter()
.filter_map(|s| s.get("podcastId").cloned())
.collect();
if podcast_ids.is_empty() {
return Ok(Json(json!({ "episodes": [], "count": 0 })));
}
let options = list_options(&params, 30, doc! { "pubDate": -1 });
let mut episodes = find_all(
&coll(&state.db, EPISODES),
doc! { "podcastId": { "$in": podcast_ids } },
options,
)
.await
.map_err(internal(ERR))?;
populate_podcasts(&state.db, &mut episodes, LIST_FIELDS)
.await
.map_err(internal(ERR))?;
let episodes = with_progress(&state, Some(&user), episodes)
.await
.map_err(internal(ERR))?;
Ok(Json(json!({ "count": episodes.len(), "episodes": episodes })))
}
pub async fn for_podcast(
State(state): State<AppState>,
MaybeUser(user): MaybeUser,
Path(podcast_id): Path<String>,
Query(params): Params,
) -> ApiResult {
const ERR: &str = "Failed to fetch episodes";
let Some(podcast_id) = db::oid(&podcast_id) else {
return Ok(Json(json!({ "episodes": [], "count": 0 })));
};
let options = list_options(&params, 50, doc! { "pubDate": -1 });
let episodes = find_all(&coll(&state.db, EPISODES), doc! { "podcastId": podcast_id }, options)
.await
.map_err(internal(ERR))?;
let episodes = with_progress(&state, user.as_ref(), episodes)
.await
.map_err(internal(ERR))?;
Ok(Json(json!({ "count": episodes.len(), "episodes": episodes })))
}
pub async fn get_one(State(state): State<AppState>, Path(id): Path<String>) -> ApiResult {
const ERR: &str = "Failed to fetch episode";
let not_found = || AppError::not_found("Episode not found");
let id = db::oid(&id).ok_or_else(not_found)?;
let episode = coll(&state.db, EPISODES)
.find_one(doc! { "_id": id }, None)
.await
.map_err(internal(ERR))?
.ok_or_else(not_found)?;
let mut episodes = vec![episode];
populate_podcasts(&state.db, &mut episodes, &["title", "author", "description", "imageUrl"])
.await
.map_err(internal(ERR))?;
let episode = episodes.pop().map(db::doc_to_json);
Ok(Json(json!({ "episode": episode })))
}
pub async fn save_progress(
State(state): State<AppState>,
user: AuthUser,
Json(body): Json<Value>,
) -> ApiResult {
let episode_id = body.get("episodeId").and_then(Value::as_str).unwrap_or_default();
let position = body.get("position").and_then(Value::as_f64);
let (Some(position), false) = (position, episode_id.is_empty()) else {
return Err(AppError::bad_request("Episode ID and position required"));
};
let episode_id = db::oid(episode_id).ok_or_else(|| AppError::bad_request("Invalid episode ID"))?;
let is_completed = body.get("isCompleted").and_then(Value::as_bool);
let now = db::now();
let mut set = doc! { "position": position.max(0.0), "lastListenedAt": now, "updatedAt": now };
let mut on_insert = doc! { "isSaved": false, "createdAt": now, "__v": 0 };
match is_completed {
Some(completed) => set.insert("isCompleted", completed),
None => on_insert.insert("isCompleted", false),
};
let options = FindOneAndUpdateOptions::builder()
.upsert(true)
.return_document(ReturnDocument::After)
.build();
let progress = coll(&state.db, PROGRESS)
.find_one_and_update(
doc! { "userId": user.id, "episodeId": episode_id },
doc! { "$set": set, "$setOnInsert": on_insert },
options,
)
.await
.map_err(internal("Failed to save progress"))?;
Ok(Json(json!({
"message": "Progress saved",
"progress": progress.map(db::doc_to_json),
})))
}
pub async fn get_progress(
State(state): State<AppState>,
user: AuthUser,
Path(episode_id): Path<String>,
) -> ApiResult {
let none = || Json(json!({ "progress": null, "message": "No progress found" }));
let Some(episode_id) = db::oid(&episode_id) else {
return Ok(none());
};
let progress = coll(&state.db, PROGRESS)
.find_one(doc! { "userId": user.id, "episodeId": episode_id }, None)
.await
.map_err(internal("Failed to fetch progress"))?;
Ok(match progress {
Some(progress) => Json(json!({ "progress": db::doc_to_json(progress) })),
None => none(),
})
}
pub async fn search(
State(state): State<AppState>,
user: AuthUser,
Query(params): Params,
) -> ApiResult {
const ERR: &str = "Search failed";
let query = params.get("q").map(String::as_str).unwrap_or_default();
if query.is_empty() {
return Err(AppError::bad_request("Search query required"));
}
let score = doc! { "score": { "$meta": "textScore" } };
let mut options = list_options(&params, 40, score.clone());
options.projection = Some(score);
let mut episodes = find_all(
&coll(&state.db, EPISODES),
doc! { "$text": { "$search": query } },
options,
)
.await
.map_err(internal(ERR))?;
populate_podcasts(&state.db, &mut episodes, LIST_FIELDS)
.await
.map_err(internal(ERR))?;
let episodes = with_progress(&state, Some(&user), episodes)
.await
.map_err(internal(ERR))?;
Ok(Json(json!({ "count": episodes.len(), "episodes": episodes })))
}
+66
View File
@@ -0,0 +1,66 @@
use axum::{
http::StatusCode,
response::{IntoResponse, Response},
Json,
};
use serde_json::{json, Value};
/// API error rendered as a JSON body (`{ "message": ... }` by default).
#[derive(Debug)]
pub struct AppError {
pub status: StatusCode,
pub body: Value,
}
impl AppError {
pub fn new(status: StatusCode, message: impl Into<String>) -> Self {
Self {
status,
body: json!({ "message": message.into() }),
}
}
pub fn with_body(status: StatusCode, body: Value) -> Self {
Self { status, body }
}
pub fn bad_request(message: impl Into<String>) -> Self {
Self::new(StatusCode::BAD_REQUEST, message)
}
pub fn unauthorized(message: impl Into<String>) -> Self {
Self::new(StatusCode::UNAUTHORIZED, message)
}
pub fn not_found(message: impl Into<String>) -> Self {
Self::new(StatusCode::NOT_FOUND, message)
}
pub fn validation(errors: Vec<Value>) -> Self {
Self::with_body(
StatusCode::BAD_REQUEST,
json!({ "message": "Validation error", "errors": errors }),
)
}
}
impl IntoResponse for AppError {
fn into_response(self) -> Response {
(self.status, Json(self.body)).into_response()
}
}
pub type ApiResult<T = Json<Value>> = Result<T, AppError>;
/// Logs the underlying error and maps it to a 500 carrying `message`.
pub fn internal<E: std::fmt::Display>(message: &'static str) -> impl FnOnce(E) -> AppError {
move |err| {
tracing::error!("{message}: {err}");
AppError::new(StatusCode::INTERNAL_SERVER_ERROR, message)
}
}
/// Builds a zod-like validation issue.
pub fn issue(field: &str, message: &str) -> Value {
json!({ "path": [field], "message": message })
}
+722
View File
@@ -0,0 +1,722 @@
//! RSS feed fetching, parsing and episode synchronisation.
use std::{collections::HashSet, time::Duration};
use anyhow::{anyhow, bail, Context};
use bson::{doc, oid::ObjectId, Bson, Document};
use chrono::{DateTime, NaiveDate, NaiveDateTime, Utc};
use mongodb::{
error::{ErrorKind, Error as MongoError},
options::{FindOptions, InsertManyOptions},
Collection,
};
use redis::AsyncCommands;
use reqwest::header::ACCEPT;
use serde_json::{json, Value};
use crate::{
db::{self, coll, EPISODES, PODCASTS},
AppState,
};
const SYNC_CACHE_TTL_SECS: u64 = 6 * 60 * 60;
#[derive(Debug)]
pub struct FeedData {
pub title: String,
pub description: String,
pub author: String,
pub image_url: Option<String>,
pub language: String,
pub episodes: Vec<ParsedEpisode>,
}
#[derive(Debug)]
pub struct ParsedEpisode {
pub title: String,
pub description: String,
pub audio_url: String,
pub pub_date: DateTime<Utc>,
pub duration: i64,
pub guid: String,
pub image_url: Option<String>,
}
pub async fn fetch_feed(http: &reqwest::Client, url: &str) -> anyhow::Result<FeedData> {
fetch_and_parse(http, url).await.map_err(|err| {
tracing::warn!("RSS parsing error for {url}: {err}");
anyhow!("Failed to parse RSS feed: {err}")
})
}
async fn fetch_and_parse(http: &reqwest::Client, url: &str) -> anyhow::Result<FeedData> {
let response = http
.get(url)
.header(ACCEPT, "application/rss+xml, application/xml, text/xml, */*")
.timeout(Duration::from_secs(30))
.send()
.await?;
if !response.status().is_success() {
bail!("Status code {}", response.status().as_u16());
}
let bytes = response.bytes().await?;
let channel = rss::Channel::read_from(&bytes[..])?;
parse_channel(&channel)
}
fn non_empty(value: Option<&str>) -> Option<&str> {
value.map(str::trim).filter(|s| !s.is_empty())
}
pub fn parse_channel(channel: &rss::Channel) -> anyhow::Result<FeedData> {
let title = channel.title().trim();
if title.is_empty() {
bail!("Invalid RSS feed - missing title");
}
let itunes = channel.itunes_ext();
let image_url = non_empty(channel.image().map(|i| i.url()))
.or_else(|| non_empty(itunes.and_then(|i| i.image())))
.map(str::to_owned);
let author = non_empty(itunes.and_then(|i| i.author()))
.or_else(|| {
non_empty(
channel
.dublin_core_ext()
.and_then(|dc| dc.creators().first().map(String::as_str)),
)
})
.or_else(|| non_empty(channel.managing_editor()))
.unwrap_or_default()
.to_owned();
let description = non_empty(Some(channel.description()))
.or_else(|| non_empty(itunes.and_then(|i| i.summary())))
.or_else(|| non_empty(itunes.and_then(|i| i.subtitle())))
.unwrap_or_default()
.to_owned();
let language = channel
.language()
.and_then(|l| l.split('-').next())
.map(|l| l.trim().to_lowercase())
.unwrap_or_default();
let mut episodes: Vec<ParsedEpisode> = channel
.items()
.iter()
.filter_map(|item| parse_item(item, image_url.as_deref()))
.collect();
episodes.sort_by(|a, b| b.pub_date.cmp(&a.pub_date));
tracing::info!(
"[RSS Parser] \"{title}\": author={author:?}, image={}, episodes={}",
image_url.is_some(),
episodes.len()
);
Ok(FeedData {
title: title.to_owned(),
description,
author,
image_url,
language,
episodes,
})
}
fn media_attr<'a>(item: &'a rss::Item, name: &str) -> Option<&'a str> {
item.extensions()
.get("media")
.and_then(|media| media.get("content"))
.and_then(|contents| contents.first())
.and_then(|ext| ext.attrs().get(name))
.map(String::as_str)
}
fn is_mp3(url: &str) -> bool {
let lower = url.to_ascii_lowercase();
lower.ends_with(".mp3") || lower.contains(".mp3?")
}
/// First `href="...mp3..."` found in an HTML fragment.
fn mp3_href(html: &str) -> Option<String> {
let mut rest = html;
while let Some(pos) = rest.find("href=") {
rest = &rest[pos + 5..];
let Some(quote) = rest.chars().next().filter(|c| *c == '"' || *c == '\'') else {
continue;
};
rest = &rest[1..];
let end = rest.find(quote)?;
let candidate = &rest[..end];
if candidate.to_ascii_lowercase().contains(".mp3") {
return Some(candidate.to_owned());
}
rest = &rest[end..];
}
None
}
fn extract_audio_url(item: &rss::Item) -> Option<String> {
if let Some(url) = non_empty(item.enclosure().map(|e| e.url())) {
return Some(url.to_owned());
}
if let Some(url) = non_empty(media_attr(item, "url")) {
return Some(url.to_owned());
}
if let Some(link) = non_empty(item.link()).filter(|l| is_mp3(l)) {
return Some(link.to_owned());
}
if let Some(guid) = non_empty(item.guid().map(|g| g.value())).filter(|g| is_mp3(g)) {
return Some(guid.to_owned());
}
item.content().and_then(mp3_href)
}
/// Parses "HH:MM:SS", "MM:SS" or a plain number of seconds.
fn parse_duration(raw: &str) -> i64 {
let parts: Option<Vec<f64>> = raw
.trim()
.split(':')
.map(|p| p.trim().parse::<f64>().ok())
.collect();
let seconds = match parts.as_deref() {
Some([h, m, s]) => h * 3600.0 + m * 60.0 + s,
Some([m, s]) => m * 60.0 + s,
Some([s]) => *s,
_ => 0.0,
};
if seconds.is_finite() && seconds > 0.0 {
seconds as i64
} else {
0
}
}
fn extract_image_url(item: &rss::Item, podcast_image: Option<&str>) -> Option<String> {
if let Some(image) = non_empty(item.itunes_ext().and_then(|i| i.image())) {
return Some(image.to_owned());
}
if media_attr(item, "medium") == Some("image") {
if let Some(url) = non_empty(media_attr(item, "url")) {
return Some(url.to_owned());
}
}
podcast_image.map(str::to_owned)
}
/// Lenient date parsing: RFC 2822 (incl. common named time zones), RFC 3339, plain dates.
pub fn parse_date(raw: &str) -> Option<DateTime<Utc>> {
let raw = raw.trim();
if let Ok(dt) = DateTime::parse_from_rfc2822(raw) {
return Some(dt.with_timezone(&Utc));
}
if let Ok(dt) = DateTime::parse_from_rfc3339(raw) {
return Some(dt.with_timezone(&Utc));
}
const ZONES: &[(&str, &str)] = &[
("CEST", "+0200"), ("CET", "+0100"), ("BST", "+0100"), ("IST", "+0530"),
("PDT", "-0700"), ("PST", "-0800"), ("MDT", "-0600"), ("MST", "-0700"),
("CDT", "-0500"), ("CST", "-0600"), ("EDT", "-0400"), ("EST", "-0500"),
("UTC", "+0000"), ("GMT", "+0000"), ("Z", "+0000"),
];
for (name, offset) in ZONES {
if let Some(prefix) = raw.strip_suffix(name) {
let candidate = format!("{} {offset}", prefix.trim_end());
if let Ok(dt) = DateTime::parse_from_rfc2822(&candidate) {
return Some(dt.with_timezone(&Utc));
}
}
}
// Drop a leading weekday ("Mon, ") that may be wrong or localised.
let without_weekday = raw.split_once(", ").map(|(_, rest)| rest).unwrap_or(raw);
for format in ["%d %b %Y %H:%M:%S %z", "%d %B %Y %H:%M:%S %z", "%Y-%m-%dT%H:%M:%S%.f%z"] {
if let Ok(dt) = DateTime::parse_from_str(without_weekday, format) {
return Some(dt.with_timezone(&Utc));
}
}
for format in ["%d %b %Y %H:%M:%S", "%d %b %Y %H:%M", "%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S"] {
if let Ok(dt) = NaiveDateTime::parse_from_str(without_weekday, format) {
return Some(dt.and_utc());
}
}
for format in ["%Y-%m-%d", "%d %b %Y"] {
if let Ok(date) = NaiveDate::parse_from_str(without_weekday, format) {
return date.and_hms_opt(0, 0, 0).map(|dt| dt.and_utc());
}
}
None
}
fn decode_entities(text: &str) -> String {
let mut out = String::with_capacity(text.len());
let mut rest = text;
while let Some(pos) = rest.find('&') {
out.push_str(&rest[..pos]);
rest = &rest[pos..];
let decoded = rest.find(';').filter(|end| *end <= 10).and_then(|end| {
let entity = &rest[1..end];
let ch = match entity {
"amp" => Some('&'),
"lt" => Some('<'),
"gt" => Some('>'),
"quot" => Some('"'),
"apos" => Some('\''),
"nbsp" => Some('\u{a0}'),
"hellip" => Some('…'),
"rsquo" => Some('’'),
"lsquo" => Some('‘'),
"rdquo" => Some('”'),
"ldquo" => Some('“'),
"ndash" => Some('–'),
"mdash" => Some('—'),
"eacute" => Some('é'),
"egrave" => Some('è'),
"agrave" => Some('à'),
"ccedil" => Some('ç'),
_ => entity
.strip_prefix("#x")
.or_else(|| entity.strip_prefix("#X"))
.and_then(|hex| u32::from_str_radix(hex, 16).ok())
.or_else(|| entity.strip_prefix('#').and_then(|d| d.parse().ok()))
.and_then(char::from_u32),
}?;
Some((ch, end))
});
match decoded {
Some((ch, end)) => {
out.push(ch);
rest = &rest[end + 1..];
}
None => {
out.push('&');
rest = &rest[1..];
}
}
}
out.push_str(rest);
out
}
/// Plain-text snippet of an HTML fragment (mirrors rss-parser's `contentSnippet`).
pub fn html_to_text(html: &str) -> String {
const BLOCK_TAGS: &[&str] = &[
"br", "p", "ul", "ol", "li", "blockquote", "section", "table", "tr", "div", "h1", "h2",
"h3", "h4", "h5", "h6",
];
let mut out = String::with_capacity(html.len());
let mut rest = html;
while let Some(start) = rest.find('<') {
out.push_str(&rest[..start]);
let Some(end) = rest[start..].find('>') else {
rest = &rest[start..];
break;
};
let tag = rest[start + 1..start + end]
.trim_start_matches('/')
.split(|c: char| c.is_whitespace() || c == '/')
.next()
.unwrap_or_default()
.to_ascii_lowercase();
if BLOCK_TAGS.contains(&tag.as_str()) && !out.ends_with('\n') && !out.is_empty() {
out.push('\n');
}
rest = &rest[start + end + 1..];
}
out.push_str(rest);
decode_entities(&out).trim().to_owned()
}
fn parse_item(item: &rss::Item, podcast_image: Option<&str>) -> Option<ParsedEpisode> {
let title = non_empty(item.title())?.to_owned();
let audio_url = extract_audio_url(item)?;
let itunes = item.itunes_ext();
let description = non_empty(item.description())
.or_else(|| non_empty(itunes.and_then(|i| i.summary())))
.or_else(|| non_empty(item.content()))
.map(html_to_text)
.unwrap_or_default();
let raw_date = non_empty(item.pub_date());
let pub_date = raw_date.and_then(parse_date).unwrap_or_else(Utc::now);
let guid = non_empty(item.guid().map(|g| g.value()))
.or_else(|| non_empty(item.link()))
.map(str::to_owned)
.unwrap_or_else(|| format!("{title}-{}", raw_date.unwrap_or("undefined")));
Some(ParsedEpisode {
description,
audio_url,
pub_date,
duration: itunes.and_then(|i| i.duration()).map(parse_duration).unwrap_or(0),
guid,
image_url: extract_image_url(item, podcast_image),
title,
})
}
fn opt_bson(value: &Option<String>) -> Bson {
value.clone().map(Bson::String).unwrap_or(Bson::Null)
}
fn episode_doc(podcast_id: ObjectId, episode: &ParsedEpisode, now: bson::DateTime) -> Document {
doc! {
"podcastId": podcast_id,
"title": episode.title.as_str(),
"description": episode.description.as_str(),
"audioUrl": episode.audio_url.as_str(),
"pubDate": bson::DateTime::from_chrono(episode.pub_date),
"duration": episode.duration,
"guid": episode.guid.as_str(),
"imageUrl": opt_bson(&episode.image_url),
"createdAt": now,
"updatedAt": now,
"__v": 0,
}
}
/// Inserts episodes, silently skipping ones already stored (duplicate guid).
async fn insert_episodes(
collection: &Collection<Document>,
docs: Vec<Document>,
) -> mongodb::error::Result<u64> {
if docs.is_empty() {
return Ok(0);
}
let total = docs.len() as u64;
let options = InsertManyOptions::builder().ordered(false).build();
match collection.insert_many(docs, options).await {
Ok(result) => Ok(result.inserted_ids.len() as u64),
Err(err) => match duplicate_only_failures(&err) {
Some(skipped) => Ok(total.saturating_sub(skipped)),
None => Err(err),
},
}
}
fn duplicate_only_failures(err: &MongoError) -> Option<u64> {
match err.kind.as_ref() {
ErrorKind::BulkWrite(failure) if failure.write_concern_error.is_none() => {
let errors = failure.write_errors.as_ref()?;
errors
.iter()
.all(|e| e.code == 11000)
.then_some(errors.len() as u64)
}
_ => None,
}
}
/// Creates a podcast (and its episodes) from an RSS feed. If another request
/// created it concurrently, the existing document is returned.
pub async fn create_podcast_from_rss(state: &AppState, rss_url: &str) -> anyhow::Result<Document> {
let feed = fetch_feed(&state.http, rss_url).await?;
let podcasts = coll(&state.db, PODCASTS);
let now = db::now();
let latest = feed.episodes.first();
let language = if feed.language.is_empty() { "en" } else { feed.language.as_str() };
let last_title = latest.map(|e| e.title.clone()).unwrap_or_default();
let last_date = latest
.map(|e| Bson::DateTime(bson::DateTime::from_chrono(e.pub_date)))
.unwrap_or(Bson::Null);
let mut podcast = doc! {
"title": feed.title.as_str(),
"description": feed.description.as_str(),
"rssUrl": rss_url,
"imageUrl": opt_bson(&feed.image_url),
"author": feed.author.as_str(),
"category": [],
"language": language,
"lastFetched": now,
"episodeCount": 0,
"lastEpisodeTitle": last_title,
"lastEpisodeDate": last_date,
"createdAt": now,
"updatedAt": now,
"__v": 0,
};
let podcast_id = match podcasts.insert_one(&podcast, None).await {
Ok(result) => result
.inserted_id
.as_object_id()
.context("unexpected podcast id type")?,
Err(err) if db::is_duplicate_key(&err) => {
return podcasts
.find_one(doc! { "rssUrl": rss_url }, None)
.await?
.ok_or_else(|| anyhow!("Podcast not found"));
}
Err(err) => return Err(err.into()),
};
let episodes = coll(&state.db, EPISODES);
let docs = feed
.episodes
.iter()
.map(|e| episode_doc(podcast_id, e, now))
.collect();
insert_episodes(&episodes, docs).await?;
let count = episodes
.count_documents(doc! { "podcastId": podcast_id }, None)
.await? as i64;
podcasts
.update_one(
doc! { "_id": podcast_id },
doc! { "$set": { "episodeCount": count } },
None,
)
.await?;
if feed.episodes.is_empty() {
tracing::warn!("[RSS Parser] No episodes found for {} during creation.", feed.title);
}
podcast.insert("_id", podcast_id);
podcast.insert("episodeCount", count);
Ok(podcast)
}
fn is_blank(podcast: &Document, key: &str) -> bool {
podcast.get_str(key).map(|s| s.trim().is_empty()).unwrap_or(true)
}
/// Fetches the feed and stores new episodes. Results are cached in Redis for
/// six hours unless `force` is set.
pub async fn sync_podcast(
state: &AppState,
podcast_id: ObjectId,
rss_url: &str,
force: bool,
) -> anyhow::Result<Value> {
let cache_key = format!("podcast:{}:sync", podcast_id.to_hex());
let mut redis = state.redis.clone();
if !force {
if let Ok(Some(cached)) = redis.get::<_, Option<String>>(&cache_key).await {
if let Ok(value) = serde_json::from_str::<Value>(&cached) {
return Ok(value);
}
}
}
let podcasts = coll(&state.db, PODCASTS);
let podcast = podcasts
.find_one(doc! { "_id": podcast_id }, None)
.await?
.ok_or_else(|| anyhow!("Podcast not found"))?;
let feed = fetch_feed(&state.http, rss_url).await?;
let episodes = coll(&state.db, EPISODES);
let now = db::now();
let known: HashSet<String> = episodes
.distinct("guid", doc! { "podcastId": podcast_id }, None)
.await?
.into_iter()
.filter_map(|guid| guid.as_str().map(str::to_owned))
.collect();
let new_docs: Vec<Document> = feed
.episodes
.iter()
.filter(|e| !known.contains(&e.guid))
.map(|e| episode_doc(podcast_id, e, now))
.collect();
let new_count = insert_episodes(&episodes, new_docs).await?;
let total = episodes
.count_documents(doc! { "podcastId": podcast_id }, None)
.await? as i64;
let mut set = doc! { "lastFetched": now, "episodeCount": total, "updatedAt": now };
if feed.title != "Unknown Podcast" {
set.insert("title", feed.title.as_str());
}
if !feed.author.is_empty() && (force || is_blank(&podcast, "author")) {
set.insert("author", feed.author.as_str());
}
if let Some(image) = &feed.image_url {
if force || is_blank(&podcast, "imageUrl") {
set.insert("imageUrl", image.as_str());
}
}
if !feed.description.is_empty() && (force || is_blank(&podcast, "description")) {
set.insert("description", feed.description.as_str());
}
if !feed.language.is_empty() && (force || is_blank(&podcast, "language")) {
set.insert("language", feed.language.as_str());
}
if let Some(latest) = feed.episodes.first() {
set.insert("lastEpisodeTitle", latest.title.as_str());
set.insert("lastEpisodeDate", bson::DateTime::from_chrono(latest.pub_date));
}
podcasts
.update_one(doc! { "_id": podcast_id }, doc! { "$set": set }, None)
.await?;
let result = json!({
"success": true,
"newEpisodes": new_count,
"totalEpisodes": total,
"lastFetched": db::iso(now),
});
if let Err(err) = redis
.set_ex::<_, _, ()>(&cache_key, result.to_string(), SYNC_CACHE_TTL_SECS)
.await
{
tracing::warn!("Redis cache write failed: {err}");
}
Ok(result)
}
async fn podcasts_matching(state: &AppState, filter: Document) -> Vec<Document> {
let options = FindOptions::builder()
.projection(doc! { "_id": 1, "rssUrl": 1, "title": 1 })
.build();
match db::find_all(&coll(&state.db, PODCASTS), filter, options).await {
Ok(podcasts) => podcasts,
Err(err) => {
tracing::error!("[Sync Engine] Failed to list podcasts: {err}");
Vec::new()
}
}
}
async fn sync_all(state: AppState) {
let podcasts = podcasts_matching(&state, doc! {}).await;
tracing::info!("[Sync Engine] Processing {} podcasts...", podcasts.len());
for podcast in podcasts {
let (Ok(id), Ok(url)) = (podcast.get_object_id("_id"), podcast.get_str("rssUrl")) else {
continue;
};
if let Err(err) = sync_podcast(&state, id, url, false).await {
let title = podcast.get_str("title").unwrap_or_default();
tracing::error!("[Sync Engine] Failed to sync {title}: {err}");
}
// Avoid hammering feed hosts.
tokio::time::sleep(Duration::from_secs(5)).await;
}
tracing::info!("[Sync Engine] Global sync completed");
}
async fn sync_empty(state: AppState) {
let podcasts = podcasts_matching(&state, doc! { "episodeCount": 0 }).await;
if podcasts.is_empty() {
return;
}
tracing::info!("[Sync Engine] Retrying {} empty podcasts...", podcasts.len());
for podcast in podcasts {
let (Ok(id), Ok(url)) = (podcast.get_object_id("_id"), podcast.get_str("rssUrl")) else {
continue;
};
if let Err(err) = sync_podcast(&state, id, url, true).await {
tracing::error!("[Sync Engine] Empty podcast sync failed: {err}");
}
}
}
/// Hourly: retry podcasts without episodes. Every 4 hours: sync the whole library.
pub fn start_background_sync(state: AppState) {
tracing::info!("✓ Background Sync Engine started");
tokio::spawn(async move {
loop {
let now = Utc::now();
let next = now.timestamp() - now.timestamp().rem_euclid(3600) + 3600;
let wait_ms = (next * 1000 - now.timestamp_millis()).max(1000) as u64;
tokio::time::sleep(Duration::from_millis(wait_ms)).await;
if (next / 3600) % 4 == 0 {
tracing::info!("[Cron] Starting global library sync");
tokio::spawn(sync_all(state.clone()));
}
tokio::spawn(sync_empty(state.clone()));
}
});
}
#[cfg(test)]
mod tests {
use super::*;
const FEED: &str = r#"<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0" xmlns:itunes="http://www.itunes.com/dtds/podcast-1.0.dtd" xmlns:media="http://search.yahoo.com/mrss/">
<channel>
<title>Mon Podcast</title>
<description>Desc</description>
<language>fr-FR</language>
<itunes:author>Radio</itunes:author>
<itunes:image href="https://example.com/cover.jpg"/>
<item>
<title>Episode 1</title>
<description><![CDATA[<p>Hello &amp; welcome</p><p>Second</p>]]></description>
<pubDate>Mon, 01 Jan 2024 10:00:00 +0000</pubDate>
<guid>ep-1</guid>
<enclosure url="https://example.com/1.mp3" type="audio/mpeg" length="1"/>
<itunes:duration>01:02:03</itunes:duration>
</item>
<item>
<title>Episode 2</title>
<pubDate>Tue, 02 Jan 2024 10:00:00 CET</pubDate>
<media:content url="https://example.com/2.mp3" medium="audio"/>
</item>
<item>
<title>No audio</title>
</item>
</channel>
</rss>"#;
#[test]
fn parses_feed() {
let channel = rss::Channel::read_from(FEED.as_bytes()).unwrap();
let feed = parse_channel(&channel).unwrap();
assert_eq!(feed.title, "Mon Podcast");
assert_eq!(feed.author, "Radio");
assert_eq!(feed.language, "fr");
assert_eq!(feed.image_url.as_deref(), Some("https://example.com/cover.jpg"));
assert_eq!(feed.episodes.len(), 2);
let newest = &feed.episodes[0];
assert_eq!(newest.title, "Episode 2");
assert_eq!(newest.audio_url, "https://example.com/2.mp3");
assert_eq!(newest.image_url.as_deref(), Some("https://example.com/cover.jpg"));
let first = &feed.episodes[1];
assert_eq!(first.guid, "ep-1");
assert_eq!(first.duration, 3723);
assert_eq!(first.description, "Hello & welcome\nSecond");
}
#[test]
fn parses_durations() {
assert_eq!(parse_duration("1:00:00"), 3600);
assert_eq!(parse_duration("05:30"), 330);
assert_eq!(parse_duration("1234"), 1234);
assert_eq!(parse_duration("abc"), 0);
}
#[test]
fn parses_dates() {
assert!(parse_date("Mon, 01 Jan 2024 10:00:00 GMT").is_some());
assert!(parse_date("Mon, 01 Jan 2024 10:00:00 CEST").is_some());
assert!(parse_date("2024-01-01T10:00:00Z").is_some());
assert!(parse_date("2024-01-01").is_some());
assert!(parse_date("not a date").is_none());
}
#[test]
fn finds_mp3_in_html() {
assert_eq!(
mp3_href(r#"<a href="https://x.com/a.mp3?x=1">dl</a>"#).as_deref(),
Some("https://x.com/a.mp3?x=1")
);
assert_eq!(mp3_href(r#"<a href=nope>x</a>"#), None);
}
}
-124
View File
@@ -1,124 +0,0 @@
import express from 'express';
import cors from 'cors';
import 'dotenv/config';
import path from 'path';
import helmet from 'helmet';
import rateLimit from 'express-rate-limit';
import { connectDB } from './config/database';
import { initRedis, closeRedis } from './config/redis';
import { syncService } from './services/syncService';
import { errorHandler } from './middleware/errorHandler';
import authRoutes from './routes/authRoutes';
import podcastRoutes from './routes/podcastRoutes';
import episodeRoutes from './routes/episodeRoutes';
import audioRoutes from './routes/audioRoutes';
const app = express();
const PORT = process.env.PORT || 5000;
// Security Middleware
app.use(helmet({
contentSecurityPolicy: false, // Disable CSP for simplicity in this development phase
}));
// Rate Limiting
const apiLimiter = rateLimit({
windowMs: 15 * 60 * 1000, // 15 minutes
max: 100, // limit each IP to 100 requests per window
message: { message: 'Trop de requêtes, veuillez réessayer plus tard.' },
standardHeaders: true,
legacyHeaders: false,
});
const discoveryLimiter = rateLimit({
windowMs: 60 * 60 * 1000, // 1 hour
max: 50, // lower limit for heavy RSS operations
message: { message: 'Limite de recherche atteinte pour cette heure.' },
});
// Middleware
app.use(cors());
app.use(express.json());
app.use(express.urlencoded({ limit: '10mb', extended: true }));
// Initialize databases on startup
let isInitialized = false;
const initialize = async () => {
if (isInitialized) return;
try {
await connectDB();
await initRedis();
isInitialized = true;
console.log('✓ All services initialized');
} catch (error) {
console.error('✗ Initialization failed:', error);
process.exit(1);
}
};
// Health check endpoint (no auth required)
app.get('/health', (req, res) => {
res.json({ status: 'ok', timestamp: new Date().toISOString() });
});
// API Routes
app.get('/api/health', (req, res) => {
res.json({ status: 'ok', message: 'API is running' });
});
app.use('/api/auth', authRoutes);
app.use('/api/podcasts', apiLimiter, podcastRoutes);
app.use('/api/episodes', apiLimiter, episodeRoutes);
app.use('/api/audio', audioRoutes);
// Serve frontend static files in production
const frontendDist = path.join(__dirname, '..', 'public');
app.use(express.static(frontendDist));
// SPA fallback — return index.html for any non-API route
app.get('*', (req, res) => {
const indexFile = path.join(frontendDist, 'index.html');
res.sendFile(indexFile, (err) => {
if (err) {
res.status(404).json({ success: false, message: 'Not found' });
}
});
});
// Error handler (must be last)
app.use(errorHandler);
// Graceful shutdown
const gracefulShutdown = async () => {
console.log('\nShutting down gracefully...');
await closeRedis();
process.exit(0);
};
process.on('SIGTERM', gracefulShutdown);
process.on('SIGINT', gracefulShutdown);
// Start server
const startServer = async () => {
try {
await initialize();
syncService.startBackgroundSync();
app.listen(PORT, () => {
console.log(`\n${'='.repeat(50)}`);
console.log(`🎙️ Podcastic Backend`);
console.log(`Server running on http://localhost:${PORT}`);
console.log(`Environment: ${process.env.NODE_ENV || 'development'}`);
console.log(`${'='.repeat(50)}\n`);
});
} catch (error) {
console.error('Failed to start server:', error);
process.exit(1);
}
};
startServer();
export default app;
+139
View File
@@ -0,0 +1,139 @@
//! iTunes Search API — free, no API key required.
use std::time::Duration;
use serde::Serialize;
use serde_json::Value;
const TIMEOUT: Duration = Duration::from_secs(10);
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct SearchResult {
pub id: String,
pub title: String,
pub rss_url: String,
pub description: String,
pub author: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub image_url: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub episode_count: Option<i64>,
}
fn str_of<'a>(item: &'a Value, key: &str) -> Option<&'a str> {
item.get(key)
.and_then(Value::as_str)
.filter(|s| !s.is_empty())
}
fn map_result(item: &Value) -> Option<SearchResult> {
let rss_url = str_of(item, "feedUrl")?;
let id = match item.get("collectionId")? {
Value::Number(n) => n.to_string(),
Value::String(s) => s.clone(),
_ => return None,
};
Some(SearchResult {
id,
title: str_of(item, "collectionName").unwrap_or_default().to_owned(),
rss_url: rss_url.to_owned(),
description: str_of(item, "description")
.or_else(|| str_of(item, "primaryGenreName"))
.unwrap_or_default()
.to_owned(),
author: str_of(item, "artistName").unwrap_or_default().to_owned(),
image_url: str_of(item, "artworkUrl600")
.or_else(|| str_of(item, "artworkUrl100"))
.map(str::to_owned),
episode_count: item.get("trackCount").and_then(Value::as_i64),
})
}
fn map_results(body: &Value) -> Vec<SearchResult> {
body.get("results")
.and_then(Value::as_array)
.map(|items| items.iter().filter_map(map_result).collect())
.unwrap_or_default()
}
pub async fn search(
http: &reqwest::Client,
query: &str,
limit: usize,
) -> anyhow::Result<Vec<SearchResult>> {
let term = query.trim();
if term.is_empty() {
return Ok(Vec::new());
}
// Fetch more candidates than needed, then respect the requested limit.
let body: Value = http
.get("https://itunes.apple.com/search")
.query(&[
("term", term),
("media", "podcast"),
("entity", "podcast"),
("country", "FR"),
("limit", "100"),
])
.timeout(TIMEOUT)
.send()
.await?
.error_for_status()?
.json()
.await?;
let mut results = map_results(&body);
results.truncate(limit);
Ok(results)
}
pub async fn trending(
http: &reqwest::Client,
limit: usize,
genre: Option<&str>,
) -> anyhow::Result<Vec<SearchResult>> {
let cap = limit.clamp(1, 50);
let genre_path = genre
.filter(|g| !g.is_empty() && g.chars().all(|c| c.is_ascii_alphanumeric()))
.map(|g| format!("/genre={g}"))
.unwrap_or_default();
let feed: Value = http
.get(format!(
"https://itunes.apple.com/fr/rss/toppodcasts/limit={cap}{genre_path}/json"
))
.timeout(TIMEOUT)
.send()
.await?
.error_for_status()?
.json()
.await?;
// A single entry is serialized as an object rather than an array.
let entries = match feed.pointer("/feed/entry") {
Some(Value::Array(entries)) => entries.clone(),
Some(entry @ Value::Object(_)) => vec![entry.clone()],
_ => Vec::new(),
};
let ids: Vec<&str> = entries
.iter()
.filter_map(|e| e.pointer("/id/attributes/im:id").and_then(Value::as_str))
.collect();
if ids.is_empty() {
return Ok(Vec::new());
}
let lookup: Value = http
.get("https://itunes.apple.com/lookup")
.query(&[("id", ids.join(",").as_str()), ("entity", "podcast")])
.timeout(TIMEOUT)
.send()
.await?
.error_for_status()?
.json()
.await?;
Ok(map_results(&lookup))
}
+237
View File
@@ -0,0 +1,237 @@
mod auth;
mod db;
mod episodes;
mod error;
mod feeds;
mod itunes;
mod podcasts;
use std::{
collections::HashMap,
env,
net::{IpAddr, SocketAddr},
path::PathBuf,
sync::{Arc, Mutex},
time::{Duration, Instant},
};
use axum::{
extract::{ConnectInfo, Request, State},
http::{header, HeaderValue, StatusCode},
middleware::{self, Next},
response::{IntoResponse, Response},
routing::{delete, get, post},
Json, Router,
};
use mongodb::Database;
use redis::aio::ConnectionManager;
use serde_json::json;
use tower_http::{
cors::CorsLayer,
services::{ServeDir, ServeFile},
set_header::SetResponseHeaderLayer,
};
use crate::auth::JwtConfig;
#[derive(Clone)]
pub struct AppState {
pub db: Database,
pub redis: ConnectionManager,
pub http: reqwest::Client,
pub jwt: Arc<JwtConfig>,
limiter: Arc<RateLimiter>,
}
/// Fixed-window, per-IP request limiter for the API routes.
struct RateLimiter {
window: Duration,
max: u32,
hits: Mutex<HashMap<IpAddr, (Instant, u32)>>,
}
impl RateLimiter {
fn new(window: Duration, max: u32) -> Self {
Self { window, max, hits: Mutex::new(HashMap::new()) }
}
fn allow(&self, ip: IpAddr) -> bool {
let now = Instant::now();
let mut hits = self.hits.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
if hits.len() > 10_000 {
hits.retain(|_, (start, _)| now.duration_since(*start) < self.window);
}
let entry = hits.entry(ip).or_insert((now, 0));
if now.duration_since(entry.0) >= self.window {
*entry = (now, 0);
}
entry.1 += 1;
entry.1 <= self.max
}
}
async fn rate_limit(
State(state): State<AppState>,
ConnectInfo(addr): ConnectInfo<SocketAddr>,
request: Request,
next: Next,
) -> Response {
if state.limiter.allow(addr.ip()) {
next.run(request).await
} else {
let body = json!({ "message": "Trop de requêtes, veuillez réessayer plus tard." });
(StatusCode::TOO_MANY_REQUESTS, Json(body)).into_response()
}
}
async fn connect_redis(url: &str) -> anyhow::Result<ConnectionManager> {
let client = redis::Client::open(url)?;
let mut attempt = 1;
loop {
match ConnectionManager::new(client.clone()).await {
Ok(manager) => {
tracing::info!("✓ Redis connected");
return Ok(manager);
}
Err(err) if attempt < 15 => {
tracing::warn!("Redis not ready (attempt {attempt}): {err}");
attempt += 1;
tokio::time::sleep(Duration::from_secs(2)).await;
}
Err(err) => return Err(err.into()),
}
}
}
fn api_router(state: &AppState) -> Router<AppState> {
let limited = || middleware::from_fn_with_state(state.clone(), rate_limit);
let auth = Router::new()
.route("/register", post(auth::register))
.route("/login", post(auth::login))
.route("/me", get(auth::me))
.route("/logout", post(auth::logout));
let podcasts = Router::new()
.route("/subscriptions", get(podcasts::subscriptions))
.route("/subscribe", post(podcasts::subscribe))
.route("/subscribe-discovery", post(podcasts::subscribe_discovery))
.route("/discover", get(podcasts::discover))
.route("/trending", get(podcasts::trending))
.route("/search", get(podcasts::search))
.route("/:id", get(podcasts::get_one))
.route("/:id/unsubscribe", delete(podcasts::unsubscribe))
.route("/:id/sync", post(podcasts::sync))
.layer(limited());
let episodes = Router::new()
.route("/latest", get(episodes::latest))
.route("/search", get(episodes::search))
.route("/progress", post(episodes::save_progress))
.route("/progress/:id", get(episodes::get_progress))
.route("/podcast/:id", get(episodes::for_podcast))
.route("/:id", get(episodes::get_one))
.layer(limited());
Router::new()
.route("/health", get(|| async { Json(json!({ "status": "ok", "message": "API is running" })) }))
.nest("/auth", auth)
.nest("/podcasts", podcasts)
.nest("/episodes", episodes)
}
#[tokio::main]
async fn main() -> anyhow::Result<()> {
tracing_subscriber::fmt()
.with_env_filter(
tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("podcastic=info,tower_http=warn")),
)
.init();
let var = |key: &str, default: &str| env::var(key).unwrap_or_else(|_| default.to_owned());
let port: u16 = var("PORT", "3579").parse()?;
let jwt_secret = var("JWT_SECRET", "dev_secret_key");
if jwt_secret == "dev_secret_key" || jwt_secret == "change_me_in_production" {
tracing::warn!("JWT_SECRET uses a default value — set a strong secret in production");
}
let db = db::connect(&var("MONGODB_URI", "mongodb://localhost:27017/podcastic")).await?;
let redis = connect_redis(&var("REDIS_URL", "redis://localhost:6379")).await?;
let http = reqwest::Client::builder()
.user_agent(
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36",
)
.connect_timeout(Duration::from_secs(10))
.redirect(reqwest::redirect::Policy::limited(10))
.build()?;
let state = AppState {
db,
redis,
http,
jwt: Arc::new(JwtConfig::new(
&jwt_secret,
&var("JWT_EXPIRES_IN", "7d"),
&var("JWT_REFRESH_EXPIRES_IN", "30d"),
)),
limiter: Arc::new(RateLimiter::new(Duration::from_secs(15 * 60), 300)),
};
feeds::start_background_sync(state.clone());
// Frontend build served as static files with an SPA fallback.
let public = PathBuf::from(var("PUBLIC_DIR", "public"));
let spa = ServeDir::new(&public).fallback(ServeFile::new(public.join("index.html")));
let app = Router::new()
.route(
"/health",
get(|| async { Json(json!({ "status": "ok", "timestamp": chrono::Utc::now().to_rfc3339() })) }),
)
.nest("/api", api_router(&state))
.fallback_service(spa)
.layer(CorsLayer::permissive())
.layer(SetResponseHeaderLayer::overriding(
header::X_CONTENT_TYPE_OPTIONS,
HeaderValue::from_static("nosniff"),
))
.layer(SetResponseHeaderLayer::overriding(
header::X_FRAME_OPTIONS,
HeaderValue::from_static("SAMEORIGIN"),
))
.layer(SetResponseHeaderLayer::overriding(
header::REFERRER_POLICY,
HeaderValue::from_static("no-referrer"),
))
.with_state(state);
let listener = tokio::net::TcpListener::bind(SocketAddr::from(([0, 0, 0, 0], port))).await?;
tracing::info!("🎙️ Podcastic (Rust) running on http://0.0.0.0:{port}");
axum::serve(listener, app.into_make_service_with_connect_info::<SocketAddr>())
.with_graceful_shutdown(shutdown_signal())
.await?;
Ok(())
}
async fn shutdown_signal() {
let ctrl_c = async {
let _ = tokio::signal::ctrl_c().await;
};
#[cfg(unix)]
let terminate = async {
if let Ok(mut signal) =
tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
{
signal.recv().await;
}
};
#[cfg(not(unix))]
let terminate = std::future::pending::<()>();
tokio::select! {
_ = ctrl_c => {},
_ = terminate => {},
}
tracing::info!("Shutting down gracefully...");
}
-54
View File
@@ -1,54 +0,0 @@
import { Request, Response, NextFunction } from 'express';
import { verifyAccessToken, JwtPayload } from '../utils/jwt';
// Extend Express Request to include user info
declare global {
namespace Express {
interface Request {
user?: JwtPayload;
}
}
}
export const authenticate = (req: Request, res: Response, next: NextFunction) => {
try {
const authHeader = req.headers.authorization;
if (!authHeader || !authHeader.startsWith('Bearer ')) {
return res.status(401).json({ message: 'Missing or invalid authorization header' });
}
const token = authHeader.substring(7); // Remove 'Bearer ' prefix
try {
const decoded = verifyAccessToken(token);
req.user = decoded;
next();
} catch (error) {
return res.status(401).json({ message: 'Invalid or expired token' });
}
} catch (error) {
return res.status(500).json({ message: 'Authentication error' });
}
};
export const optional = (req: Request, res: Response, next: NextFunction) => {
try {
const authHeader = req.headers.authorization;
if (authHeader && authHeader.startsWith('Bearer ')) {
const token = authHeader.substring(7);
try {
const decoded = verifyAccessToken(token);
req.user = decoded;
} catch (error) {
// Token invalid but optional, continue
}
}
next();
} catch (error) {
next();
}
};
-31
View File
@@ -1,31 +0,0 @@
import { Request, Response, NextFunction } from 'express';
export interface AppError extends Error {
statusCode?: number;
details?: unknown;
}
export const errorHandler = (
error: AppError,
req: Request,
res: Response,
next: NextFunction
) => {
const statusCode = error.statusCode || 500;
const message = error.message || 'Internal Server Error';
console.error(`[${statusCode}] ${message}`, error.details || '');
res.status(statusCode).json({
success: false,
message,
...(process.env.NODE_ENV === 'development' && { details: error.details }),
});
};
export const notFoundHandler = (req: Request, res: Response) => {
res.status(404).json({
success: false,
message: 'Route not found',
});
};
-65
View File
@@ -1,65 +0,0 @@
import mongoose, { Schema, Document } from 'mongoose';
export interface IEpisode extends Document {
podcastId: mongoose.Types.ObjectId;
title: string;
description: string;
audioUrl: string;
pubDate: Date;
duration: number; // in seconds
guid: string; // unique identifier from RSS
imageUrl?: string;
createdAt: Date;
updatedAt: Date;
}
const episodeSchema = new Schema<IEpisode>(
{
podcastId: {
type: Schema.Types.ObjectId,
ref: 'Podcast',
required: true,
index: true,
},
title: {
type: String,
required: true,
trim: true,
},
description: {
type: String,
default: '',
},
audioUrl: {
type: String,
required: true,
},
pubDate: {
type: Date,
required: true,
index: true,
},
duration: {
type: Number,
default: 0,
},
guid: {
type: String,
required: true,
},
imageUrl: {
type: String,
default: null,
},
},
{
timestamps: true,
}
);
// Compound index for unique episode per podcast
episodeSchema.index({ podcastId: 1, guid: 1 }, { unique: true });
episodeSchema.index({ pubDate: -1 });
episodeSchema.index({ title: 'text', description: 'text' });
export const Episode = mongoose.model<IEpisode>('Episode', episodeSchema);
-79
View File
@@ -1,79 +0,0 @@
import mongoose, { Schema, Document } from 'mongoose';
export interface IPodcast extends Document {
title: string;
description: string;
rssUrl: string;
imageUrl?: string;
author: string;
category: string[];
language: string;
lastFetched: Date;
episodeCount: number;
lastEpisodeTitle?: string;
lastEpisodeDate?: Date;
createdAt: Date;
updatedAt: Date;
}
const podcastSchema = new Schema<IPodcast>(
{
title: {
type: String,
required: true,
trim: true,
},
description: {
type: String,
default: '',
},
rssUrl: {
type: String,
required: true,
unique: true,
trim: true,
},
imageUrl: {
type: String,
default: null,
},
author: {
type: String,
trim: true,
},
category: {
type: [String],
default: [],
},
language: {
type: String,
default: 'en',
},
lastFetched: {
type: Date,
default: () => new Date(0), // Start with epoch
},
episodeCount: {
type: Number,
default: 0,
},
lastEpisodeTitle: {
type: String,
default: '',
},
lastEpisodeDate: {
type: Date,
default: null,
},
},
{
timestamps: true,
}
);
// Index for faster queries
podcastSchema.index({ rssUrl: 1 });
podcastSchema.index({ title: 'text', description: 'text', author: 'text' });
podcastSchema.index({ category: 1 });
export const Podcast = mongoose.model<IPodcast>('Podcast', podcastSchema);
-89
View File
@@ -1,89 +0,0 @@
import mongoose, { Schema, Document } from 'mongoose';
import bcrypt from 'bcryptjs';
export interface IUser extends Document {
email: string;
password: string;
username: string;
avatar?: string;
preferences: {
defaultPlaybackSpeed: number;
theme: 'light' | 'dark';
language: string;
};
createdAt: Date;
updatedAt: Date;
comparePassword(password: string): Promise<boolean>;
}
const userSchema = new Schema<IUser>(
{
email: {
type: String,
required: true,
unique: true,
lowercase: true,
trim: true,
},
password: {
type: String,
required: true,
minlength: 6,
},
username: {
type: String,
required: true,
trim: true,
},
avatar: {
type: String,
default: null,
},
preferences: {
defaultPlaybackSpeed: {
type: Number,
default: 1,
enum: [0.75, 1, 1.25, 1.5, 1.75, 2],
},
theme: {
type: String,
default: 'light',
enum: ['light', 'dark'],
},
language: {
type: String,
default: 'en',
},
},
},
{
timestamps: true,
}
);
// Hash password before saving
userSchema.pre('save', async function (next) {
if (!this.isModified('password')) return next();
try {
const salt = await bcrypt.genSalt(10);
this.password = await bcrypt.hash(this.password, salt);
next();
} catch (error) {
next(error as Error);
}
});
// Method to compare passwords
userSchema.methods.comparePassword = async function (password: string): Promise<boolean> {
return bcrypt.compare(password, this.password);
};
// Remove password from JSON output
userSchema.methods.toJSON = function () {
const obj = this.toObject();
delete obj.password;
return obj;
};
export const User = mongoose.model<IUser>('User', userSchema);
-57
View File
@@ -1,57 +0,0 @@
import mongoose, { Schema, Document } from 'mongoose';
export interface IUserProgress extends Document {
userId: mongoose.Types.ObjectId;
episodeId: mongoose.Types.ObjectId;
position: number; // seconds listened
isCompleted: boolean;
isSaved: boolean;
lastListenedAt: Date;
createdAt: Date;
updatedAt: Date;
}
const userProgressSchema = new Schema<IUserProgress>(
{
userId: {
type: Schema.Types.ObjectId,
ref: 'User',
required: true,
index: true,
},
episodeId: {
type: Schema.Types.ObjectId,
ref: 'Episode',
required: true,
index: true,
},
position: {
type: Number,
default: 0,
min: 0,
},
isCompleted: {
type: Boolean,
default: false,
},
isSaved: {
type: Boolean,
default: false,
},
lastListenedAt: {
type: Date,
default: () => new Date(),
},
},
{
timestamps: true,
}
);
// Compound index for unique progress per user per episode
userProgressSchema.index({ userId: 1, episodeId: 1 }, { unique: true });
userProgressSchema.index({ userId: 1, lastListenedAt: -1 });
userProgressSchema.index({ userId: 1, isCompleted: 1 });
userProgressSchema.index({ userId: 1, isSaved: 1 });
export const UserProgress = mongoose.model<IUserProgress>('UserProgress', userProgressSchema);
-47
View File
@@ -1,47 +0,0 @@
import mongoose, { Schema, Document } from 'mongoose';
export interface IUserSubscription extends Document {
userId: mongoose.Types.ObjectId;
podcastId: mongoose.Types.ObjectId;
subscribedAt: Date;
isFavorite: boolean;
createdAt: Date;
updatedAt: Date;
}
const userSubscriptionSchema = new Schema<IUserSubscription>(
{
userId: {
type: Schema.Types.ObjectId,
ref: 'User',
required: true,
index: true,
},
podcastId: {
type: Schema.Types.ObjectId,
ref: 'Podcast',
required: true,
index: true,
},
subscribedAt: {
type: Date,
default: () => new Date(),
},
isFavorite: {
type: Boolean,
default: false,
},
},
{
timestamps: true,
}
);
// Compound index to ensure unique subscription per user
userSubscriptionSchema.index({ userId: 1, podcastId: 1 }, { unique: true });
userSubscriptionSchema.index({ userId: 1, isFavorite: 1 });
export const UserSubscription = mongoose.model<IUserSubscription>(
'UserSubscription',
userSubscriptionSchema
);
+311
View File
@@ -0,0 +1,311 @@
use std::collections::HashMap;
use axum::{
extract::{Path, Query, State},
http::StatusCode,
Json,
};
use bson::{doc, oid::ObjectId, Document};
use mongodb::options::FindOptions;
use serde_json::{json, Value};
use crate::{
auth::AuthUser,
db::{self, coll, find_all, PODCASTS, SUBSCRIPTIONS},
error::{internal, issue, ApiResult, AppError},
feeds, itunes, AppState,
};
type Params = Query<HashMap<String, String>>;
/// `Number(value) || default` semantics for pagination parameters.
pub fn num_param(params: &HashMap<String, String>, key: &str, default: i64) -> i64 {
params
.get(key)
.and_then(|v| v.trim().parse::<f64>().ok())
.filter(|v| v.is_finite() && *v > 0.0)
.map(|v| v as i64)
.unwrap_or(default)
}
fn escape_regex(text: &str) -> String {
let mut out = String::with_capacity(text.len());
for c in text.chars() {
if "\\^$.|?*+()[]{}/".contains(c) {
out.push('\\');
}
out.push(c);
}
out
}
pub async fn subscriptions(State(state): State<AppState>, user: AuthUser) -> ApiResult {
let options = FindOptions::builder().sort(doc! { "subscribedAt": -1 }).build();
let subs = find_all(&coll(&state.db, SUBSCRIPTIONS), doc! { "userId": user.id }, options)
.await
.map_err(internal("Failed to fetch subscriptions"))?;
let ids: Vec<ObjectId> = subs
.iter()
.filter_map(|s| s.get_object_id("podcastId").ok())
.collect();
let mut podcasts: HashMap<ObjectId, Document> =
find_all(&coll(&state.db, PODCASTS), doc! { "_id": { "$in": ids } }, None)
.await
.map_err(internal("Failed to fetch subscriptions"))?
.into_iter()
.filter_map(|p| Some((p.get_object_id("_id").ok()?, p)))
.collect();
let list: Vec<Value> = subs
.iter()
.filter_map(|sub| {
let mut podcast = podcasts.remove(&sub.get_object_id("podcastId").ok()?)?;
podcast.insert("subscriptionId", sub.get_object_id("_id").ok()?.to_hex());
Some(db::doc_to_json(podcast))
})
.collect();
Ok(Json(json!({ "count": list.len(), "podcasts": list })))
}
async fn subscribe_user(state: &AppState, user: &AuthUser, rss_url: &str) -> ApiResult {
let podcasts = coll(&state.db, PODCASTS);
let existing = podcasts
.find_one(doc! { "rssUrl": rss_url }, None)
.await
.map_err(internal("Impossible de s'abonner"))?;
let podcast = match existing {
Some(podcast) => podcast,
None => feeds::create_podcast_from_rss(state, rss_url)
.await
.map_err(|err| {
AppError::bad_request(format!("Impossible de récupérer le podcast : {err}"))
})?,
};
let podcast_id = podcast
.get_object_id("_id")
.map_err(internal("Impossible de s'abonner"))?;
let subs = coll(&state.db, SUBSCRIPTIONS);
let filter = doc! { "userId": user.id, "podcastId": podcast_id };
let already = subs
.find_one(filter, None)
.await
.map_err(internal("Impossible de s'abonner"))?
.is_some();
if !already {
let now = db::now();
let subscription = doc! {
"userId": user.id,
"podcastId": podcast_id,
"subscribedAt": now,
"isFavorite": false,
"createdAt": now,
"updatedAt": now,
"__v": 0,
};
if let Err(err) = subs.insert_one(subscription, None).await {
if !db::is_duplicate_key(&err) {
return Err(internal("Impossible de s'abonner")(err));
}
}
}
Ok(Json(json!({
"message": if already { "Déjà abonné" } else { "Abonnement réussi" },
"podcast": db::doc_to_json(podcast),
})))
}
pub async fn subscribe(
State(state): State<AppState>,
user: AuthUser,
Json(body): Json<Value>,
) -> ApiResult {
let rss_url = body
.get("rssUrl")
.and_then(Value::as_str)
.map(str::trim)
.unwrap_or_default();
if reqwest::Url::parse(rss_url).is_err() {
return Err(AppError::validation(vec![issue("rssUrl", "Invalid RSS URL")]));
}
subscribe_user(&state, &user, rss_url).await
}
pub async fn subscribe_discovery(
State(state): State<AppState>,
user: AuthUser,
Json(body): Json<Value>,
) -> ApiResult {
let rss_url = body
.get("rssUrl")
.and_then(Value::as_str)
.map(str::trim)
.unwrap_or_default();
if rss_url.is_empty() {
return Err(AppError::bad_request("RSS URL required"));
}
subscribe_user(&state, &user, rss_url).await
}
pub async fn unsubscribe(
State(state): State<AppState>,
user: AuthUser,
Path(id): Path<String>,
) -> ApiResult {
let not_found = || AppError::not_found("Subscription not found");
let id = db::oid(&id).ok_or_else(not_found)?;
let subs = coll(&state.db, SUBSCRIPTIONS);
// The frontend may send either the subscription id or the podcast id.
let mut deleted = subs
.find_one_and_delete(doc! { "_id": id, "userId": user.id }, None)
.await
.map_err(internal("Failed to unsubscribe"))?;
if deleted.is_none() {
deleted = subs
.find_one_and_delete(doc! { "userId": user.id, "podcastId": id }, None)
.await
.map_err(internal("Failed to unsubscribe"))?;
}
let deleted = deleted.ok_or_else(not_found)?;
let deleted_id = deleted.get_object_id("_id").map(|id| id.to_hex()).unwrap_or_default();
tracing::info!("[Unsubscribe] Removed subscription {deleted_id}");
Ok(Json(json!({ "message": "Unsubscribed successfully", "deletedId": deleted_id })))
}
pub async fn sync(
State(state): State<AppState>,
_user: AuthUser,
Path(id): Path<String>,
) -> ApiResult {
let not_found = || AppError::not_found("Podcast not found");
let id = db::oid(&id).ok_or_else(not_found)?;
let podcast = coll(&state.db, PODCASTS)
.find_one(doc! { "_id": id }, None)
.await
.map_err(internal("Failed to sync podcast"))?
.ok_or_else(not_found)?;
let rss_url = podcast.get_str("rssUrl").unwrap_or_default();
tracing::info!("[Sync] Manual sync for {}", podcast.get_str("title").unwrap_or_default());
feeds::sync_podcast(&state, id, rss_url, true)
.await
.map(Json)
.map_err(|err| {
tracing::error!("Manual sync error: {err}");
AppError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
})
}
pub async fn search(State(state): State<AppState>, Query(params): Params) -> ApiResult {
let query = params.get("q").map(String::as_str).unwrap_or_default();
if query.is_empty() {
return Err(AppError::bad_request("Search query required"));
}
let score = doc! { "score": { "$meta": "textScore" } };
let options = FindOptions::builder()
.projection(score.clone())
.sort(score)
.skip(num_param(&params, "skip", 0) as u64)
.limit(num_param(&params, "limit", 20))
.build();
let podcasts: Vec<Value> = find_all(
&coll(&state.db, PODCASTS),
doc! { "$text": { "$search": query } },
options,
)
.await
.map_err(internal("Search failed"))?
.into_iter()
.map(db::doc_to_json)
.collect();
Ok(Json(json!({ "count": podcasts.len(), "podcasts": podcasts })))
}
pub async fn get_one(State(state): State<AppState>, Path(id): Path<String>) -> ApiResult {
let not_found = || AppError::not_found("Podcast not found");
let id = db::oid(&id).ok_or_else(not_found)?;
let podcast = coll(&state.db, PODCASTS)
.find_one(doc! { "_id": id }, None)
.await
.map_err(internal("Failed to fetch podcast"))?
.ok_or_else(not_found)?;
Ok(Json(json!({ "podcast": db::doc_to_json(podcast) })))
}
fn empty_discovery() -> Json<Value> {
Json(json!({ "source": "itunes", "podcasts": [], "count": 0 }))
}
pub async fn discover(State(state): State<AppState>, Query(params): Params) -> Json<Value> {
let query = params.get("q").map(String::as_str).unwrap_or_default();
let limit = match params.get("limit").map(|l| l.trim().parse::<i64>()) {
None => 20,
Some(Ok(limit)) if (1..=50).contains(&limit) => limit,
Some(_) => return empty_discovery(),
};
if query.chars().count() < 2 {
return empty_discovery();
}
match itunes::search(&state.http, query, limit as usize).await {
Ok(results) if !results.is_empty() => {
return Json(json!({ "source": "itunes", "count": results.len(), "podcasts": results }));
}
Ok(_) => tracing::info!("[Discover] iTunes returned 0 results for \"{query}\", trying local DB"),
Err(err) => tracing::error!("[Discover] iTunes failed for \"{query}\": {err}"),
}
// Fallback: podcasts already known locally.
let pattern = escape_regex(query);
let filter = doc! { "$or": [
{ "title": { "$regex": pattern.as_str(), "$options": "i" } },
{ "author": { "$regex": pattern.as_str(), "$options": "i" } },
{ "description": { "$regex": pattern.as_str(), "$options": "i" } },
] };
let options = FindOptions::builder().limit(limit).build();
let local = match find_all(&coll(&state.db, PODCASTS), filter, options).await {
Ok(local) => local,
Err(err) => {
tracing::error!("Discover podcasts error: {err}");
return empty_discovery();
}
};
let mapped: Vec<Value> = local
.into_iter()
.map(|p| {
let p = db::doc_to_json(p);
json!({
"id": p["_id"],
"title": p["title"],
"rssUrl": p["rssUrl"],
"description": p.get("description").cloned().unwrap_or_else(|| json!("")),
"author": p.get("author").cloned().unwrap_or_else(|| json!("")),
"imageUrl": p["imageUrl"],
"episodeCount": p["episodeCount"],
})
})
.collect();
Json(json!({ "source": "local", "count": mapped.len(), "podcasts": mapped }))
}
pub async fn trending(State(state): State<AppState>, Query(params): Params) -> Json<Value> {
let limit = num_param(&params, "limit", 20) as usize;
let genre = params.get("genre").map(String::as_str);
match itunes::trending(&state.http, limit, genre).await {
Ok(podcasts) => Json(json!({ "source": "itunes", "count": podcasts.len(), "podcasts": podcasts })),
Err(err) => {
tracing::error!("Trending podcasts error: {err}");
empty_discovery()
}
}
}
-141
View File
@@ -1,141 +0,0 @@
import { Router, Request, Response } from 'express';
import https from 'https';
import http from 'http';
import { IncomingMessage } from 'http';
const router = Router();
const BLOCKED = [
/^127\./,
/^10\./,
/^192\.168\./,
/^172\.(1[6-9]|2\d|3[01])\./,
/^::1$/,
/^localhost$/i,
];
function isBlocked(hostname: string): boolean {
return BLOCKED.some(r => r.test(hostname));
}
// Keep-alive agents reuse TLS connections across requests
const httpsAgent = new https.Agent({ keepAlive: true, maxSockets: 32 });
const httpAgent = new http.Agent({ keepAlive: true, maxSockets: 32 });
// Cache resolved final URLs (skip redirect chain on range/seek requests)
// TTL: 10 minutes
const urlCache = new Map<string, { finalUrl: string; expiresAt: number }>();
const URL_CACHE_TTL = 10 * 60 * 1000;
function getCachedUrl(original: string): string | null {
const entry = urlCache.get(original);
if (!entry) return null;
if (Date.now() > entry.expiresAt) { urlCache.delete(original); return null; }
return entry.finalUrl;
}
function setCachedUrl(original: string, finalUrl: string): void {
urlCache.set(original, { finalUrl, expiresAt: Date.now() + URL_CACHE_TTL });
}
const REDIRECT_CODES = new Set([301, 302, 303, 307, 308]);
function fetchAudio(
url: string,
originalUrl: string,
headers: Record<string, string>,
res: Response,
depth: number,
): void {
if (depth > 8) {
if (!res.headersSent) res.status(502).json({ message: 'Too many redirects' });
return;
}
let parsed: URL;
try {
parsed = new URL(url);
} catch {
if (!res.headersSent) res.status(400).json({ message: 'Invalid redirect URL' });
return;
}
if (!['http:', 'https:'].includes(parsed.protocol)) {
if (!res.headersSent) res.status(400).json({ message: 'Protocol not allowed' });
return;
}
if (isBlocked(parsed.hostname)) {
if (!res.headersSent) res.status(403).json({ message: 'Forbidden' });
return;
}
const agent = parsed.protocol === 'https:' ? httpsAgent : httpAgent;
const lib = parsed.protocol === 'https:' ? https : http;
const proxyReq = lib.request(url, { headers, agent }, (proxyRes: IncomingMessage) => {
const status = proxyRes.statusCode ?? 200;
if (REDIRECT_CODES.has(status) && proxyRes.headers.location) {
const next = new URL(proxyRes.headers.location, url).toString();
proxyRes.resume();
fetchAudio(next, originalUrl, headers, res, depth + 1);
return;
}
// Cache resolved URL so next range request skips redirect chain
if (depth > 0 && originalUrl !== url) {
setCachedUrl(originalUrl, url);
}
res.setHeader('Access-Control-Allow-Origin', '*');
res.setHeader('Access-Control-Expose-Headers', 'Content-Length, Content-Range, Content-Type, Accept-Ranges');
const forward = ['content-type', 'content-length', 'content-range', 'accept-ranges'];
for (const h of forward) {
const v = proxyRes.headers[h];
if (v) res.setHeader(h, v as string);
}
res.status(status);
proxyRes.pipe(res);
});
proxyReq.on('error', () => {
if (!res.headersSent) res.status(502).json({ message: 'Proxy error' });
});
proxyReq.end();
}
router.get('/proxy', (req: Request, res: Response) => {
const raw = req.query.url as string;
if (!raw) { res.status(400).json({ message: 'url required' }); return; }
let parsed: URL;
try {
parsed = new URL(raw);
} catch {
res.status(400).json({ message: 'Invalid URL' }); return;
}
if (!['http:', 'https:'].includes(parsed.protocol)) {
res.status(400).json({ message: 'Protocol not allowed' }); return;
}
if (isBlocked(parsed.hostname)) {
res.status(403).json({ message: 'Forbidden' }); return;
}
const upstream: Record<string, string> = {
'User-Agent': 'Mozilla/5.0 (compatible; Podcastic/1.0)',
'Accept': '*/*',
};
if (req.headers.range) upstream['Range'] = req.headers.range as string;
req.on('close', () => res.destroy());
// Use cached final URL if available (skips redirect chain — critical for seeks)
const startUrl = getCachedUrl(raw) ?? raw;
fetchAudio(startUrl, raw, upstream, res, 0);
});
export default router;
-15
View File
@@ -1,15 +0,0 @@
import { Router } from 'express';
import { register, login, getMe, logout } from '../controllers/authController';
import { authenticate } from '../middleware/auth';
const router = Router();
// Public routes
router.post('/register', register);
router.post('/login', login);
// Protected routes
router.get('/me', authenticate, getMe);
router.post('/logout', authenticate, logout);
export default router;
-24
View File
@@ -1,24 +0,0 @@
import { Router } from 'express';
import {
getLatestEpisodes,
getPodcastEpisodes,
getEpisode,
saveProgress,
getProgress,
searchEpisodes,
} from '../controllers/episodeController';
import { authenticate } from '../middleware/auth';
const router = Router();
// Protected routes
router.get('/latest', authenticate, getLatestEpisodes);
router.get('/search', authenticate, searchEpisodes);
router.post('/progress', authenticate, saveProgress);
router.get('/progress/:episodeId', authenticate, getProgress);
// Public routes
router.get('/:episodeId', getEpisode);
router.get('/podcast/:podcastId', getPodcastEpisodes);
export default router;
-30
View File
@@ -1,30 +0,0 @@
import { Router } from 'express';
import {
getUserSubscriptions,
subscribe,
unsubscribe,
syncPodcast,
searchPodcasts,
getPodcast,
discoverPodcasts,
subscribeFromDiscovery,
getTrendingPodcasts,
} from '../controllers/podcastController';
import { authenticate } from '../middleware/auth';
const router = Router();
// Protected routes
router.get('/subscriptions', authenticate, getUserSubscriptions);
router.post('/subscribe', authenticate, subscribe);
router.post('/subscribe-discovery', authenticate, subscribeFromDiscovery);
router.delete('/:podcastId/unsubscribe', authenticate, unsubscribe);
router.post('/:podcastId/sync', authenticate, syncPodcast);
// Public/discovery routes
router.get('/discover', discoverPodcasts);
router.get('/trending', getTrendingPodcasts);
router.get('/search', searchPodcasts);
router.get('/:podcastId', getPodcast);
export default router;
-119
View File
@@ -1,119 +0,0 @@
import axios from 'axios';
/**
* iTunes Search API Service - Free, no API key required
* https://developer.apple.com/library/archive/documentation/AudioVideo/Conceptual/iTuneSearchAPI/
*/
interface ItunesSearchResult {
collectionId: number;
collectionName: string;
feedUrl?: string;
artworkUrl600?: string;
artworkUrl100?: string;
artistName?: string;
primaryGenreName?: string;
trackCount?: number;
country?: string;
description?: string;
}
interface ItunesSearchResponse {
resultCount: number;
results: ItunesSearchResult[];
}
interface ItunesTopResult {
id: string;
name: string;
artworkUrl100: string;
artistName: string;
}
export interface SearchResult {
id: string;
title: string;
rssUrl: string;
description: string;
author?: string;
imageUrl?: string;
language?: string;
episodeCount?: number;
}
const mapItunesResult = (item: ItunesSearchResult): SearchResult | null => {
if (!item.feedUrl) return null;
return {
id: item.collectionId.toString(),
title: item.collectionName,
rssUrl: item.feedUrl,
description: item.description || item.primaryGenreName || '',
author: item.artistName || '',
imageUrl: item.artworkUrl600 || item.artworkUrl100,
episodeCount: item.trackCount,
};
};
export const podcastIndexService = {
searchPodcasts: async (
query: string,
limit: number = 20,
): Promise<SearchResult[]> => {
const term = query.trim();
if (!term) return [];
const response = await axios.get<ItunesSearchResponse>(
'https://itunes.apple.com/search',
{
params: {
term,
media: 'podcast',
entity: 'podcast',
country: 'FR',
limit: 100, // Fetch more candidates to improve quality
},
timeout: 10000,
}
);
return response.data.results
.map(mapItunesResult)
.filter((r): r is SearchResult => r !== null)
.slice(0, limit); // Respect requested limit after fetching candidates
},
getTrendingPodcasts: async (limit: number = 20, genreId?: string): Promise<SearchResult[]> => {
// iTunes official top podcasts RSS
// Format: https://itunes.apple.com/fr/rss/toppodcasts/limit=X/genre=Y/json
const cap = Math.min(limit, 50);
const genrePath = genreId ? `/genre=${genreId}` : '';
const rssResponse = await axios.get<{
feed: {
entry: Array<{
id: { label: string; attributes: { 'im:id': string } };
'im:name': { label: string };
'im:artist': { label: string };
'im:image': Array<{ label: string; attributes: { height: string } }>;
}>;
};
}>(`https://itunes.apple.com/fr/rss/toppodcasts/limit=${cap}${genrePath}/json`, {
timeout: 10000,
});
const entries = rssResponse.data.feed.entry;
if (!entries || entries.length === 0) return [];
// Batch lookup to get feedUrls
const ids = entries.map((e) => e.id.attributes['im:id']).join(',');
const lookupResponse = await axios.get<ItunesSearchResponse>(
'https://itunes.apple.com/lookup',
{ params: { id: ids, entity: 'podcast' }, timeout: 10000 }
);
return lookupResponse.data.results
.map(mapItunesResult)
.filter((r): r is SearchResult => r !== null);
},
isConfigured: (): boolean => true,
};
-368
View File
@@ -1,368 +0,0 @@
import Parser from 'rss-parser';
import { Podcast } from '../models/Podcast';
import { Episode } from '../models/Episode';
import { getRedis } from '../config/redis';
const parser = new Parser({
timeout: 30000, // Increased to 30 seconds for complex feeds
headers: {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36',
'Accept': 'application/rss+xml, application/xml, text/xml, */*'
},
customFields: {
item: [
['media:content', 'mediaContent'],
['itunes:summary', 'itunesSummary'],
['itunes:subtitle', 'itunesSubtitle'],
['itunes:duration', 'itunesDuration'],
['enclosure', 'enclosure'],
],
},
});
interface ParsedEpisode {
title: string;
description: string;
audioUrl: string;
pubDate: Date;
duration: number;
guid: string;
imageUrl?: string;
}
/**
* Parse RSS feed and extract podcast & episode data
*/
export const rssParserService = {
/**
* Parse RSS feed from URL
*/
parseFeed: async (rssUrl: string) => {
try {
const feed = await parser.parseURL(rssUrl);
if (!feed.title) {
throw new Error('Invalid RSS feed - missing title');
}
// Robust image extraction
let imageUrl = feed.image?.url;
if (!imageUrl && feed.itunes?.image) {
imageUrl = typeof feed.itunes.image === 'string' ? feed.itunes.image : (feed.itunes.image as any).href;
}
// Robust author extraction
const author = feed.author || feed.creator || feed.itunes?.author || (feed as any).managingEditor || '';
// Robust description extraction
const description = feed.description || feed.itunes?.summary || feed.itunes?.subtitle || (feed as any).content || '';
console.log(`[RSS Parser] Meta Extracted for "${feed.title}": Author: ${author}, Image: ${!!imageUrl}`);
return {
title: feed.title,
description: description,
author: author,
imageUrl: imageUrl,
language: feed.language?.split('-')[0].toLowerCase() || '',
episodes: feed.items || [],
};
} catch (error: any) {
console.error('RSS parsing error:', error.message);
throw new Error(`Failed to parse RSS feed: ${error.message}`);
}
},
/**
* Extract audio URL from episode item
*/
extractAudioUrl: (item: any): string | null => {
// Try enclosure first (standard RSS)
if (item.enclosure?.url) {
return item.enclosure.url;
}
// Try media:content
if (item.mediaContent?.url) {
return item.mediaContent.url;
}
// Try finding mp3 in link or guid
const mp3Regex = /\.mp3($|\?)/i;
if (item.link && mp3Regex.test(item.link)) {
return item.link;
}
if (item.guid && mp3Regex.test(item.guid)) {
return item.guid;
}
// Try finding mp3 in content:encoded
if (item['content:encoded']) {
const match = item['content:encoded'].match(/href=["']([^"']*\.mp3[^"']*)["']/);
if (match) return match[1];
}
return null;
},
/**
* Extract episode duration in seconds
*/
extractDuration: (item: any): number => {
// Try iTunes duration
if (item.itunes?.duration) {
const duration = item.itunes.duration;
// Handle "HH:MM:SS" or "MM:SS" or seconds format
if (typeof duration === 'string') {
const parts = duration.split(':').map(Number);
if (parts.length === 3) {
return parts[0] * 3600 + parts[1] * 60 + parts[2];
} else if (parts.length === 2) {
return parts[0] * 60 + parts[1];
} else if (parts.length === 1) {
return parts[0];
}
}
if (typeof duration === 'number') {
return duration;
}
}
return 0;
},
/**
* Extract episode image URL
*/
extractImageUrl: (item: any, podcastImage?: string): string | undefined => {
// Try iTunes image first
if (item.itunes?.image) {
return typeof item.itunes.image === 'string' ? item.itunes.image : (item.itunes.image as any).href;
}
// Try media content
if (item.mediaContent?.medium === 'image') {
return item.mediaContent.url;
}
// Fall back to podcast image
return podcastImage;
},
/**
* Parse episodes from RSS feed items
*/
parseEpisodes: (items: any[], podcastImage?: string, rssUrl: string = ''): ParsedEpisode[] => {
const parsed = items
.filter((item) => {
const audioUrl = rssParserService.extractAudioUrl(item);
const hasTitle = !!item.title;
if (!audioUrl || !hasTitle) {
// Log only a sample to avoid flooding, but enough to debug
if (items.indexOf(item) === 0) {
console.log(`[RSS Debug] Item check failed for "${item.title || 'Untitled'}": Audio: ${!!audioUrl}, RSS URL Sample: ${rssUrl.substring(0, 30)}...`);
if (!audioUrl) console.log(`[RSS Debug] Available keys for item: ${Object.keys(item).join(', ')}`);
}
}
return !!audioUrl && hasTitle;
});
return parsed.map((item) => ({
title: item.title || 'Untitled Episode',
description: item.contentSnippet || item.description || '',
audioUrl: rssParserService.extractAudioUrl(item) || '',
pubDate: new Date(item.pubDate || Date.now()),
duration: rssParserService.extractDuration(item),
guid: item.guid || item.link || `${item.title}-${item.pubDate}`,
imageUrl: rssParserService.extractImageUrl(item, podcastImage),
}))
.sort((a, b) => b.pubDate.getTime() - a.pubDate.getTime()); // Sort newest first
},
/**
* Fetch and sync podcast episodes
*/
syncPodcastEpisodes: async (podcastId: string, rssUrl: string, force: boolean = false) => {
try {
const redis = getRedis();
const cacheKey = `podcast:${podcastId}:sync`;
// Check cache unless forced
if (!force) {
const cached = await redis.get(cacheKey);
if (cached) {
console.log(`[RSS Sync] Returning cached result for ${podcastId}`);
return JSON.parse(cached);
}
} else {
console.log(`[RSS Sync] Force sync requested for ${podcastId}`);
}
const podcast = await Podcast.findById(podcastId);
if (!podcast) {
throw new Error('Podcast not found');
}
// Parse RSS feed
const feedData = await rssParserService.parseFeed(rssUrl);
const episodes = rssParserService.parseEpisodes(feedData.episodes, feedData.imageUrl);
// Upsert episodes
const createdCount = await Promise.all(
episodes.map(async (episodeData) => {
const existing = await Episode.findOne({
podcastId,
guid: episodeData.guid,
});
if (!existing) {
const episode = new Episode({
podcastId,
title: episodeData.title,
description: episodeData.description,
audioUrl: episodeData.audioUrl,
pubDate: episodeData.pubDate,
duration: episodeData.duration,
guid: episodeData.guid,
imageUrl: episodeData.imageUrl,
});
await episode.save();
return 1;
}
return 0;
})
);
const newCount = createdCount.filter((c) => c === 1).length;
// Update podcast metadata
podcast.lastFetched = new Date();
podcast.episodeCount = await Episode.countDocuments({ podcastId });
// Refresh podcast metadata from fresh RSS data (fixes "Unknown Podcast" legacy records)
if (feedData.title && feedData.title !== 'Unknown Podcast') {
podcast.title = feedData.title;
}
if (feedData.author && (!podcast.author || podcast.author.trim() === '' || force)) {
podcast.author = feedData.author;
}
if (feedData.imageUrl && (!podcast.imageUrl || force)) {
podcast.imageUrl = feedData.imageUrl;
}
if (feedData.description && (!podcast.description || force)) {
podcast.description = feedData.description;
}
if (feedData.language && (!podcast.language || force)) {
podcast.language = feedData.language;
}
if (episodes.length > 0) {
podcast.lastEpisodeTitle = episodes[0].title;
podcast.lastEpisodeDate = episodes[0].pubDate;
}
await podcast.save();
const result = {
success: true,
newEpisodes: newCount,
totalEpisodes: podcast.episodeCount,
lastFetched: podcast.lastFetched,
};
// Cache for 6 hours
await redis.setEx(cacheKey, 6 * 60 * 60, JSON.stringify(result));
return result;
} catch (error: any) {
console.error('Podcast sync error:', error.message);
throw error;
}
},
/**
* Create new podcast from RSS feed
*/
createPodcastFromRss: async (
rssUrl: string,
feedData?: any
): Promise<{
podcast: any;
episodes: number;
}> => {
try {
// Check if podcast already exists
let podcast = await Podcast.findOne({ rssUrl });
if (podcast) {
throw new Error('Podcast already in database');
}
// Always parse feed to get fresh description and episodes, even if partial data provided
if (!feedData || !feedData.description || !feedData.episodes || feedData.episodes.length === 0) {
const freshData = await rssParserService.parseFeed(rssUrl);
feedData = { ...feedData, ...freshData };
}
// Create podcast
podcast = new Podcast({
title: feedData.title,
description: feedData.description,
rssUrl,
imageUrl: feedData.imageUrl,
author: feedData.author,
language: feedData.language || 'en',
category: [],
episodeCount: 0,
});
await podcast.save();
// Parse and create episodes
const episodes = rssParserService.parseEpisodes(feedData.episodes, feedData.imageUrl, rssUrl);
for (const episodeData of episodes) {
const episode = new Episode({
podcastId: podcast._id,
title: episodeData.title,
description: episodeData.description,
audioUrl: episodeData.audioUrl,
pubDate: episodeData.pubDate,
duration: episodeData.duration,
guid: episodeData.guid,
imageUrl: episodeData.imageUrl,
});
try {
await episode.save();
} catch (error) {
// Skip duplicate episodes
console.log(`Skipping duplicate episode: ${episodeData.guid}`);
}
}
podcast.episodeCount = await Episode.countDocuments({ podcastId: podcast._id });
podcast.lastFetched = new Date();
if (episodes.length > 0) {
console.log(`[RSS Parser] Setting last episode for ${podcast.title}: ${episodes[0].title} (${episodes[0].pubDate})`);
podcast.lastEpisodeTitle = episodes[0].title;
podcast.lastEpisodeDate = episodes[0].pubDate;
} else {
console.warn(`[RSS Parser] No episodes found for ${podcast.title} during creation.`);
}
await podcast.save();
return {
podcast: podcast.toJSON(),
episodes: podcast.episodeCount,
};
} catch (error: any) {
console.error('Create podcast error:', error.message);
throw error;
}
},
};
-69
View File
@@ -1,69 +0,0 @@
import cron from 'node-cron';
import { Podcast } from '../models/Podcast';
import { rssParserService } from './rssParserService';
/**
* Service to handle background synchronization of podcast feeds
*/
export const syncService = {
/**
* Initialize background sync jobs
*/
startBackgroundSync: () => {
console.log('✓ Background Sync Engine started');
// Run every 4 hours
// '0 */4 * * *'
cron.schedule('0 */4 * * *', async () => {
console.log(`[Cron] Starting global library sync: ${new Date().toISOString()}`);
await syncService.syncAllPodcasts();
});
// Run a quick sync for "Zero Episode" podcasts every hour
// (Helpful for new additions that failed initial sync)
cron.schedule('0 * * * *', async () => {
await syncService.syncEmptyPodcasts();
});
},
/**
* Sync all podcasts in the database
*/
syncAllPodcasts: async () => {
try {
const podcasts = await Podcast.find({});
console.log(`[Sync Engine] Processing ${podcasts.length} podcasts...`);
for (const podcast of podcasts) {
try {
await rssParserService.syncPodcastEpisodes(podcast._id.toString(), podcast.rssUrl);
// Wait 5 seconds between podcasts to avoid rate limiting from sources
await new Promise(resolve => setTimeout(resolve, 5000));
} catch (error: any) {
console.error(`[Sync Engine] Failed to sync ${podcast.title}:`, error.message);
}
}
console.log('[Sync Engine] Global sync completed');
} catch (error: any) {
console.error('[Sync Engine] Master sync error:', error.message);
}
},
/**
* Specifically target podcasts with 0 episodes
*/
syncEmptyPodcasts: async () => {
try {
const emptyPodcasts = await Podcast.find({ episodeCount: 0 });
if (emptyPodcasts.length === 0) return;
console.log(`[Sync Engine] Retrying ${emptyPodcasts.length} empty podcasts...`);
for (const podcast of emptyPodcasts) {
await rssParserService.syncPodcastEpisodes(podcast._id.toString(), podcast.rssUrl, true);
}
} catch (error: any) {
console.error('[Sync Engine] Empty podcasts sync error:', error.message);
}
}
};
-59
View File
@@ -1,59 +0,0 @@
import jwt from 'jsonwebtoken';
import { IUser } from '../models/User';
import type { SignOptions } from 'jsonwebtoken';
const JWT_SECRET = process.env.JWT_SECRET || 'dev_secret_key';
const JWT_EXPIRES_IN = (process.env.JWT_EXPIRES_IN || '7d') as SignOptions['expiresIn'];
const JWT_REFRESH_EXPIRES_IN = (process.env.JWT_REFRESH_EXPIRES_IN || '30d') as SignOptions['expiresIn'];
export interface JwtPayload {
id: string;
email: string;
}
export interface TokenPair {
accessToken: string;
refreshToken: string;
}
export const generateTokens = (user: IUser): TokenPair => {
const payload: JwtPayload = {
id: user._id.toString(),
email: user.email,
};
const accessToken = jwt.sign(payload, JWT_SECRET, {
expiresIn: JWT_EXPIRES_IN,
});
const refreshToken = jwt.sign(payload, JWT_SECRET, {
expiresIn: JWT_REFRESH_EXPIRES_IN,
});
return { accessToken, refreshToken };
};
export const verifyAccessToken = (token: string): JwtPayload => {
try {
return jwt.verify(token, JWT_SECRET) as JwtPayload;
} catch (error) {
throw new Error('Invalid or expired token');
}
};
export const verifyRefreshToken = (token: string): JwtPayload => {
try {
return jwt.verify(token, JWT_SECRET) as JwtPayload;
} catch (error) {
throw new Error('Invalid or expired refresh token');
}
};
export const decodeToken = (token: string): JwtPayload | null => {
try {
return jwt.decode(token) as JwtPayload | null;
} catch {
return null;
}
};
-21
View File
@@ -1,21 +0,0 @@
{
"compilerOptions": {
"target": "ES2020",
"module": "CommonJS",
"lib": ["ES2020"],
"moduleResolution": "node",
"resolveJsonModule": true,
"allowJs": true,
"strict": true,
"esModuleInterop": true,
"skipLibCheck": true,
"forceConsistentCasingInFileNames": true,
"outDir": "./dist",
"rootDir": "./src",
"declaration": true,
"declarationMap": true,
"sourceMap": true
},
"include": ["src/**/*"],
"exclude": ["node_modules", "dist"]
}
+2 -2
View File
@@ -21,9 +21,9 @@ echo ⚠️ Make sure MongoDB and Redis are running!
echo.
REM Start backend
echo 🚀 Starting Backend (Express.js)...
echo 🚀 Starting Backend (Rust)...
cd backend
start "Podcastic Backend" npm run dev
start "Podcastic Backend" cmd /c "set PORT=5000&& cargo run"
cd ..
timeout /t 2 /nobreak
+1 -1
View File
@@ -42,7 +42,7 @@ echo ""
# Start backend in background
echo "→ Backend starting on port 5000..."
cd backend
npm run dev &
PORT=5000 PUBLIC_DIR=../frontend/dist cargo run &
BACKEND_PID=$!
cd ..
+3 -1
View File
@@ -30,12 +30,14 @@ services:
app:
image: ghcr.io/r0m1k3/podcastic:latest
# Always fetch the newest image when the stack (re)starts
pull_policy: always
container_name: podcastic-app
restart: unless-stopped
ports:
- "${PORT:-3579}:3579"
environment:
NODE_ENV: production
RUST_LOG: podcastic=info,tower_http=warn
PORT: 3579
MONGODB_URI: mongodb://mongodb:27017/podcastic
REDIS_URL: redis://redis:6379