Start work on Monitor Album

Cargo.toml

@@ -27,5 +27,15 @@ tonic-health = "0.14.6"
 toml = "1.1.2"
 reqwest = { version = "0.13.4", default-features = false, features = ["query", "rustls"] }
 quick-xml = { version = "0.41.0", features = ["serialize"] }
+regex = "1"
+sha1 = "0.10"
+tracing = "0.1"
+tracing-subscriber = "0.3"
+librqbit = { version = "8", default-features = false, features = ["rust-tls"] }
+tempfile = "3"
+anyhow = "1"
+sqlx = { version = "0.9", default-features = false, features = ["postgres", "runtime-tokio", "tls-rustls", "json", "uuid"] }
+serde_json = "1"
+uuid = { version = "1", features = ["v4"] }
 
 [workspace]

config.toml

@@ -0,0 +1,11 @@
+[service]
+address = "[::1]:50051"
+[indexer]
+address = "http://localhost:9117"
+api = "pj47116b3clxpf4h1b2fu9vrj07pmfa6"
+[torrent]
+address = "unix:///run/user/1000/devenv-ae542bd/torad.sock"
+[metadata]
+address = "http://localhost:50053"
+[database]
+url = "postgres://fujin@localhost:5432/music_agregator"

db/schema.sql

@@ -0,0 +1,64 @@
+-- Cached parsed-torrent metadata. Mirrors the Rust `ParsedTorrent` struct
+-- field-for-field. A UUID `id` is the PK so rows can be inserted during the
+-- fast path (title-only, no info_hash yet) and updated when the background
+-- resolver finishes.
+--
+-- A row exists in one of three states:
+--   'partial'  - title-derived fields only, awaiting background resolution
+--   'resolving'- a background resolver task is currently fetching torrent bytes
+--   'resolved' - full torrent-derived data filled in
+CREATE TABLE parsed_torrents (
+    id                  UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    info_hash           TEXT NOT NULL DEFAULT '',
+
+    raw_title           TEXT NOT NULL DEFAULT '',
+
+    artist              TEXT NOT NULL DEFAULT '',
+    album               TEXT NOT NULL DEFAULT '',
+    year                INTEGER NOT NULL DEFAULT 0,
+
+    release_type        TEXT NOT NULL DEFAULT 'unknown',
+    genres              JSONB NOT NULL DEFAULT '[]'::jsonb,
+    label               TEXT NOT NULL DEFAULT '',
+    source              TEXT NOT NULL DEFAULT 'unknown',
+    rip_type            TEXT NOT NULL DEFAULT '',
+
+    format              TEXT NOT NULL DEFAULT 'unknown',
+    bitrate             TEXT NOT NULL DEFAULT '',
+    bit_depth           INTEGER NOT NULL DEFAULT 0,
+    sample_rate         INTEGER NOT NULL DEFAULT 0,
+
+    track_names         JSONB NOT NULL DEFAULT '[]'::jsonb,
+    track_count         INTEGER NOT NULL DEFAULT 0,
+    release_count       INTEGER NOT NULL DEFAULT 0,
+    audio_file_count    INTEGER NOT NULL DEFAULT 0,
+    total_audio_size    BIGINT NOT NULL DEFAULT 0,
+
+    has_cover_art       BOOLEAN NOT NULL DEFAULT false,
+    has_cue_sheet       BOOLEAN NOT NULL DEFAULT false,
+    has_rip_log         BOOLEAN NOT NULL DEFAULT false,
+
+    parsed_successfully BOOLEAN NOT NULL DEFAULT false,
+    parse_errors        JSONB NOT NULL DEFAULT '[]'::jsonb,
+
+    -- cache-control columns (not part of ParsedTorrent)
+    state               TEXT NOT NULL DEFAULT 'partial',
+    indexer             TEXT NOT NULL DEFAULT '',
+    magnet              TEXT NOT NULL DEFAULT '',
+    created_at          TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at          TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+-- enum-backed fields (release_type, source, format) are stored as their
+-- `as_str()` repr in plain TEXT rather than native ENUMs, so adding a variant
+-- to the Rust enum needs no migration.
+
+-- lets the background resolver find rows still awaiting torrent resolution
+CREATE INDEX parsed_torrents_state_idx
+    ON parsed_torrents (state)
+    WHERE state IN ('partial', 'resolving');
+
+-- dedup resolved torrents by info_hash (only meaningful once non-empty)
+CREATE UNIQUE INDEX parsed_torrents_info_hash_uidx
+    ON parsed_torrents (info_hash)
+    WHERE info_hash != '';

dev/Music Agregator/Jackett/Search.yml

@@ -5,10 +5,10 @@ info:
 
 http:
   method: GET
-  url: http://localhost:9117/api/v2.0/indexers/all/results/torznab/api?apikey=bjivq1zp3bfq7eaorhsswkfhvpc5issh&t=search&q=ДДТ
+  url: http://localhost:9117/api/v2.0/indexers/all/results/torznab/api?apikey=pj47116b3clxpf4h1b2fu9vrj07pmfa6&t=search&q=ДДТ
   params:
     - name: apikey
-      value: bjivq1zp3bfq7eaorhsswkfhvpc5issh
+      value: pj47116b3clxpf4h1b2fu9vrj07pmfa6
       type: query
     - name: t
       value: search

dev/Music Agregator/Montitor Album.yml

@@ -0,0 +1,16 @@
+info:
+  name: Montitor Album
+  type: grpc
+  seq: 5
+
+grpc:
+  url: localhost:50051
+  method: /music.v1.MusicService/MonitorAlbum
+  methodType: unary
+  message: |-
+    {
+      "album": "Творчество в пустоте",
+      "artist": "ДДТ",
+      "quality": 0
+    }
+  auth: inherit

dev/Music Agregator/Torrent/Search.yml

@@ -9,6 +9,6 @@ grpc:
   methodType: unary
   message: |-
     {
-      "name": "ДДТ"
+      "name": "ДДТ Творчество в пустоте"
     }
   auth: inherit

devenv.nix

@@ -44,11 +44,11 @@
 
   services.postgres = {
     enable = true;
-    # initialDatabases = [
-    #   {
-    #     name = "toradb";
-    #     schema = ./db/schema.sql;
-    #   }
-    # ];
+    initialDatabases = [
+      {
+        name = "music-agregator";
+        schema = ./db/schema.sql;
+      }
+    ];
   };
 }

proto/music/v1/music.proto

@@ -0,0 +1,22 @@
+syntax = "proto3";
+
+package music.v1;
+
+service MusicService {
+  rpc MonitorAlbum(MonitorAlbumRequest) returns (MonitorAlbumResponse);
+}
+
+enum Quality {
+  MP3 = 0;
+  FLAC = 1;
+}
+
+message MonitorAlbumRequest {
+  string album = 1;
+  string artist = 2;
+  Quality quality = 3;
+}
+
+message MonitorAlbumResponse {
+
+}

src/config.rs

@@ -6,6 +6,7 @@ pub struct Config {
     pub indexer: IndexerConfig,
     pub torrent: TorrentConfig,
     pub metadata: MetadataConfig,
+    pub database: DatabaseConfig,
 }
 
 #[derive(Debug, Deserialize)]
@@ -29,3 +30,8 @@ pub struct TorrentConfig {
 pub struct MetadataConfig {
     pub address: String,
 }
+
+#[derive(Debug, Deserialize)]
+pub struct DatabaseConfig {
+    pub url: String,
+}

src/db.rs

@@ -0,0 +1,186 @@
+use anyhow::Result;
+use sqlx::Row;
+use sqlx::postgres::{PgPool, PgPoolOptions};
+use uuid::Uuid;
+
+use crate::torrent_parser::{AudioFormat, ParsedTorrent, ReleaseSource, ReleaseType};
+
+const SCHEMA_SQL: &str = include_str!("../db/schema.sql");
+
+pub async fn connect(database_url: &str) -> Result<PgPool> {
+    let pool = PgPoolOptions::new()
+        .max_connections(5)
+        .connect(database_url)
+        .await?;
+    Ok(pool)
+}
+
+pub async fn apply_schema(pool: &PgPool) -> Result<()> {
+    sqlx::raw_sql(SCHEMA_SQL).execute(pool).await?;
+    Ok(())
+}
+
+pub async fn insert_partial(pool: &PgPool, pt: &ParsedTorrent, source_url: &str) -> Result<Uuid> {
+    let row = sqlx::query(
+        "INSERT INTO parsed_torrents (
+            info_hash, raw_title, artist, album, year,
+            release_type, genres, label, source, rip_type,
+            format, bitrate, bit_depth, sample_rate,
+            track_names, track_count, release_count, audio_file_count, total_audio_size,
+            has_cover_art, has_cue_sheet, has_rip_log,
+            parsed_successfully, parse_errors,
+            state, magnet
+        ) VALUES (
+            $1, $2, $3, $4, $5,
+            $6, $7, $8, $9, $10,
+            $11, $12, $13, $14,
+            $15, $16, $17, $18, $19,
+            $20, $21, $22,
+            $23, $24,
+            'partial', $25
+        )
+        RETURNING id",
+    )
+    .bind(&pt.info_hash)
+    .bind(&pt.raw_title)
+    .bind(&pt.artist)
+    .bind(&pt.album)
+    .bind(pt.year as i32)
+    .bind(pt.release_type.as_str())
+    .bind(sqlx::types::Json(&pt.genres))
+    .bind(&pt.label)
+    .bind(pt.source.as_str())
+    .bind(&pt.rip_type)
+    .bind(pt.format.as_str())
+    .bind(&pt.bitrate)
+    .bind(pt.bit_depth as i32)
+    .bind(pt.sample_rate as i32)
+    .bind(sqlx::types::Json(&pt.track_names))
+    .bind(pt.track_count as i32)
+    .bind(pt.release_count as i32)
+    .bind(pt.audio_file_count as i32)
+    .bind(pt.total_audio_size as i64)
+    .bind(pt.has_cover_art)
+    .bind(pt.has_cue_sheet)
+    .bind(pt.has_rip_log)
+    .bind(pt.parsed_successfully)
+    .bind(sqlx::types::Json(&pt.parse_errors))
+    .bind(source_url)
+    .fetch_one(pool)
+    .await?;
+
+    Ok(row.get("id"))
+}
+
+pub async fn update_resolved(pool: &PgPool, id: Uuid, pt: &ParsedTorrent) -> Result<()> {
+    sqlx::query(
+        "UPDATE parsed_torrents SET
+            info_hash = $2,
+            raw_title = $3,
+            artist = $4,
+            album = $5,
+            year = $6,
+            release_type = $7,
+            genres = $8,
+            label = $9,
+            source = $10,
+            rip_type = $11,
+            format = $12,
+            bitrate = $13,
+            bit_depth = $14,
+            sample_rate = $15,
+            track_names = $16,
+            track_count = $17,
+            release_count = $18,
+            audio_file_count = $19,
+            total_audio_size = $20,
+            has_cover_art = $21,
+            has_cue_sheet = $22,
+            has_rip_log = $23,
+            parsed_successfully = $24,
+            parse_errors = $25,
+            state = 'resolved',
+            updated_at = now()
+        WHERE id = $1",
+    )
+    .bind(id)
+    .bind(&pt.info_hash)
+    .bind(&pt.raw_title)
+    .bind(&pt.artist)
+    .bind(&pt.album)
+    .bind(pt.year as i32)
+    .bind(pt.release_type.as_str())
+    .bind(sqlx::types::Json(&pt.genres))
+    .bind(&pt.label)
+    .bind(pt.source.as_str())
+    .bind(&pt.rip_type)
+    .bind(pt.format.as_str())
+    .bind(&pt.bitrate)
+    .bind(pt.bit_depth as i32)
+    .bind(pt.sample_rate as i32)
+    .bind(sqlx::types::Json(&pt.track_names))
+    .bind(pt.track_count as i32)
+    .bind(pt.release_count as i32)
+    .bind(pt.audio_file_count as i32)
+    .bind(pt.total_audio_size as i64)
+    .bind(pt.has_cover_art)
+    .bind(pt.has_cue_sheet)
+    .bind(pt.has_rip_log)
+    .bind(pt.parsed_successfully)
+    .bind(sqlx::types::Json(&pt.parse_errors))
+    .execute(pool)
+    .await?;
+    Ok(())
+}
+
+pub async fn get(pool: &PgPool, info_hash: &str) -> Result<Option<ParsedTorrent>> {
+    let row = sqlx::query(
+        "SELECT info_hash, raw_title, artist, album, year,
+                release_type, genres, label, source, rip_type,
+                format, bitrate, bit_depth, sample_rate,
+                track_names, track_count, release_count, audio_file_count, total_audio_size,
+                has_cover_art, has_cue_sheet, has_rip_log,
+                parsed_successfully, parse_errors
+         FROM parsed_torrents
+         WHERE info_hash = $1 AND state = 'resolved'",
+    )
+    .bind(info_hash)
+    .fetch_optional(pool)
+    .await?;
+
+    let row = match row {
+        Some(r) => r,
+        None => return Ok(None),
+    };
+
+    let genres: sqlx::types::Json<Vec<String>> = row.try_get("genres")?;
+    let track_names: sqlx::types::Json<Vec<String>> = row.try_get("track_names")?;
+    let parse_errors: sqlx::types::Json<Vec<String>> = row.try_get("parse_errors")?;
+
+    Ok(Some(ParsedTorrent {
+        info_hash: row.try_get("info_hash")?,
+        raw_title: row.try_get("raw_title")?,
+        artist: row.try_get("artist")?,
+        album: row.try_get("album")?,
+        year: row.try_get::<i32, _>("year")? as u32,
+        release_type: ReleaseType::from_str(row.try_get("release_type")?),
+        genres: genres.0,
+        label: row.try_get("label")?,
+        source: ReleaseSource::from_str(row.try_get("source")?),
+        rip_type: row.try_get("rip_type")?,
+        format: AudioFormat::from_str(row.try_get("format")?),
+        bitrate: row.try_get("bitrate")?,
+        bit_depth: row.try_get::<i32, _>("bit_depth")? as u32,
+        sample_rate: row.try_get::<i32, _>("sample_rate")? as u32,
+        track_names: track_names.0,
+        track_count: row.try_get::<i32, _>("track_count")? as u32,
+        release_count: row.try_get::<i32, _>("release_count")? as u32,
+        audio_file_count: row.try_get::<i32, _>("audio_file_count")? as u32,
+        total_audio_size: row.try_get::<i64, _>("total_audio_size")? as u64,
+        has_cover_art: row.try_get("has_cover_art")?,
+        has_cue_sheet: row.try_get("has_cue_sheet")?,
+        has_rip_log: row.try_get("has_rip_log")?,
+        parsed_successfully: row.try_get("parsed_successfully")?,
+        parse_errors: parse_errors.0,
+    }))
+}

src/generated/music/v1/music.v1.rs

@@ -0,0 +1,111 @@
+// @generated
+// This file is @generated by prost-build.
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct MonitorAlbumRequest {
+    #[prost(string, tag = "1")]
+    pub album: ::prost::alloc::string::String,
+    #[prost(string, tag = "2")]
+    pub artist: ::prost::alloc::string::String,
+    #[prost(enumeration = "Quality", tag = "3")]
+    pub quality: i32,
+}
+impl ::prost::Name for MonitorAlbumRequest {
+    const NAME: &'static str = "MonitorAlbumRequest";
+    const PACKAGE: &'static str = "music.v1";
+    fn full_name() -> ::prost::alloc::string::String {
+        "music.v1.MonitorAlbumRequest".into()
+    }
+    fn type_url() -> ::prost::alloc::string::String {
+        "/music.v1.MonitorAlbumRequest".into()
+    }
+}
+#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct MonitorAlbumResponse {}
+impl ::prost::Name for MonitorAlbumResponse {
+    const NAME: &'static str = "MonitorAlbumResponse";
+    const PACKAGE: &'static str = "music.v1";
+    fn full_name() -> ::prost::alloc::string::String {
+        "music.v1.MonitorAlbumResponse".into()
+    }
+    fn type_url() -> ::prost::alloc::string::String {
+        "/music.v1.MonitorAlbumResponse".into()
+    }
+}
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
+#[repr(i32)]
+pub enum Quality {
+    Mp3 = 0,
+    Flac = 1,
+}
+impl Quality {
+    /// String value of the enum field names used in the ProtoBuf definition.
+    ///
+    /// The values are not transformed in any way and thus are considered stable
+    /// (if the ProtoBuf definition does not change) and safe for programmatic use.
+    pub fn as_str_name(&self) -> &'static str {
+        match self {
+            Self::Mp3 => "MP3",
+            Self::Flac => "FLAC",
+        }
+    }
+    /// Creates an enum from field names used in the ProtoBuf definition.
+    pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
+        match value {
+            "MP3" => Some(Self::Mp3),
+            "FLAC" => Some(Self::Flac),
+            _ => None,
+        }
+    }
+}
+/// Encoded file descriptor set for the `music.v1` package
+pub const FILE_DESCRIPTOR_SET: &[u8] = &[
+    0x0a, 0xdc, 0x05, 0x0a, 0x14, 0x6d, 0x75, 0x73, 0x69, 0x63, 0x2f, 0x76, 0x31, 0x2f, 0x6d, 0x75,
+    0x73, 0x69, 0x63, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x08, 0x6d, 0x75, 0x73, 0x69, 0x63,
+    0x2e, 0x76, 0x31, 0x22, 0x70, 0x0a, 0x13, 0x4d, 0x6f, 0x6e, 0x69, 0x74, 0x6f, 0x72, 0x41, 0x6c,
+    0x62, 0x75, 0x6d, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x14, 0x0a, 0x05, 0x61, 0x6c,
+    0x62, 0x75, 0x6d, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x61, 0x6c, 0x62, 0x75, 0x6d,
+    0x12, 0x16, 0x0a, 0x06, 0x61, 0x72, 0x74, 0x69, 0x73, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09,
+    0x52, 0x06, 0x61, 0x72, 0x74, 0x69, 0x73, 0x74, 0x12, 0x2b, 0x0a, 0x07, 0x71, 0x75, 0x61, 0x6c,
+    0x69, 0x74, 0x79, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x11, 0x2e, 0x6d, 0x75, 0x73, 0x69,
+    0x63, 0x2e, 0x76, 0x31, 0x2e, 0x51, 0x75, 0x61, 0x6c, 0x69, 0x74, 0x79, 0x52, 0x07, 0x71, 0x75,
+    0x61, 0x6c, 0x69, 0x74, 0x79, 0x22, 0x16, 0x0a, 0x14, 0x4d, 0x6f, 0x6e, 0x69, 0x74, 0x6f, 0x72,
+    0x41, 0x6c, 0x62, 0x75, 0x6d, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x2a, 0x1c, 0x0a,
+    0x07, 0x51, 0x75, 0x61, 0x6c, 0x69, 0x74, 0x79, 0x12, 0x07, 0x0a, 0x03, 0x4d, 0x50, 0x33, 0x10,
+    0x00, 0x12, 0x08, 0x0a, 0x04, 0x46, 0x4c, 0x41, 0x43, 0x10, 0x01, 0x32, 0x5d, 0x0a, 0x0c, 0x4d,
+    0x75, 0x73, 0x69, 0x63, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x4d, 0x0a, 0x0c, 0x4d,
+    0x6f, 0x6e, 0x69, 0x74, 0x6f, 0x72, 0x41, 0x6c, 0x62, 0x75, 0x6d, 0x12, 0x1d, 0x2e, 0x6d, 0x75,
+    0x73, 0x69, 0x63, 0x2e, 0x76, 0x31, 0x2e, 0x4d, 0x6f, 0x6e, 0x69, 0x74, 0x6f, 0x72, 0x41, 0x6c,
+    0x62, 0x75, 0x6d, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1e, 0x2e, 0x6d, 0x75, 0x73,
+    0x69, 0x63, 0x2e, 0x76, 0x31, 0x2e, 0x4d, 0x6f, 0x6e, 0x69, 0x74, 0x6f, 0x72, 0x41, 0x6c, 0x62,
+    0x75, 0x6d, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x4a, 0xaa, 0x03, 0x0a, 0x06, 0x12,
+    0x04, 0x00, 0x00, 0x15, 0x01, 0x0a, 0x08, 0x0a, 0x01, 0x0c, 0x12, 0x03, 0x00, 0x00, 0x12, 0x0a,
+    0x08, 0x0a, 0x01, 0x02, 0x12, 0x03, 0x02, 0x00, 0x11, 0x0a, 0x0a, 0x0a, 0x02, 0x06, 0x00, 0x12,
+    0x04, 0x04, 0x00, 0x06, 0x01, 0x0a, 0x0a, 0x0a, 0x03, 0x06, 0x00, 0x01, 0x12, 0x03, 0x04, 0x08,
+    0x14, 0x0a, 0x0b, 0x0a, 0x04, 0x06, 0x00, 0x02, 0x00, 0x12, 0x03, 0x05, 0x02, 0x47, 0x0a, 0x0c,
+    0x0a, 0x05, 0x06, 0x00, 0x02, 0x00, 0x01, 0x12, 0x03, 0x05, 0x06, 0x12, 0x0a, 0x0c, 0x0a, 0x05,
+    0x06, 0x00, 0x02, 0x00, 0x02, 0x12, 0x03, 0x05, 0x13, 0x26, 0x0a, 0x0c, 0x0a, 0x05, 0x06, 0x00,
+    0x02, 0x00, 0x03, 0x12, 0x03, 0x05, 0x31, 0x45, 0x0a, 0x0a, 0x0a, 0x02, 0x05, 0x00, 0x12, 0x04,
+    0x08, 0x00, 0x0b, 0x01, 0x0a, 0x0a, 0x0a, 0x03, 0x05, 0x00, 0x01, 0x12, 0x03, 0x08, 0x05, 0x0c,
+    0x0a, 0x0c, 0x0a, 0x05, 0x05, 0x00, 0x02, 0x00, 0x01, 0x12, 0x03, 0x09, 0x02, 0x05, 0x0a, 0x0b,
+    0x0a, 0x04, 0x05, 0x00, 0x02, 0x00, 0x12, 0x03, 0x09, 0x02, 0x0a, 0x0a, 0x0c, 0x0a, 0x05, 0x05,
+    0x00, 0x02, 0x00, 0x02, 0x12, 0x03, 0x09, 0x08, 0x09, 0x0a, 0x0c, 0x0a, 0x05, 0x05, 0x00, 0x02,
+    0x01, 0x01, 0x12, 0x03, 0x0a, 0x02, 0x06, 0x0a, 0x0b, 0x0a, 0x04, 0x05, 0x00, 0x02, 0x01, 0x12,
+    0x03, 0x0a, 0x02, 0x0b, 0x0a, 0x0c, 0x0a, 0x05, 0x05, 0x00, 0x02, 0x01, 0x02, 0x12, 0x03, 0x0a,
+    0x09, 0x0a, 0x0a, 0x0a, 0x0a, 0x02, 0x04, 0x00, 0x12, 0x04, 0x0d, 0x00, 0x11, 0x01, 0x0a, 0x0a,
+    0x0a, 0x03, 0x04, 0x00, 0x01, 0x12, 0x03, 0x0d, 0x08, 0x1b, 0x0a, 0x0c, 0x0a, 0x05, 0x04, 0x00,
+    0x02, 0x00, 0x05, 0x12, 0x03, 0x0e, 0x02, 0x08, 0x0a, 0x0b, 0x0a, 0x04, 0x04, 0x00, 0x02, 0x00,
+    0x12, 0x03, 0x0e, 0x02, 0x13, 0x0a, 0x0c, 0x0a, 0x05, 0x04, 0x00, 0x02, 0x00, 0x01, 0x12, 0x03,
+    0x0e, 0x09, 0x0e, 0x0a, 0x0c, 0x0a, 0x05, 0x04, 0x00, 0x02, 0x00, 0x03, 0x12, 0x03, 0x0e, 0x11,
+    0x12, 0x0a, 0x0c, 0x0a, 0x05, 0x04, 0x00, 0x02, 0x01, 0x05, 0x12, 0x03, 0x0f, 0x02, 0x08, 0x0a,
+    0x0b, 0x0a, 0x04, 0x04, 0x00, 0x02, 0x01, 0x12, 0x03, 0x0f, 0x02, 0x14, 0x0a, 0x0c, 0x0a, 0x05,
+    0x04, 0x00, 0x02, 0x01, 0x01, 0x12, 0x03, 0x0f, 0x09, 0x0f, 0x0a, 0x0c, 0x0a, 0x05, 0x04, 0x00,
+    0x02, 0x01, 0x03, 0x12, 0x03, 0x0f, 0x12, 0x13, 0x0a, 0x0c, 0x0a, 0x05, 0x04, 0x00, 0x02, 0x02,
+    0x06, 0x12, 0x03, 0x10, 0x02, 0x09, 0x0a, 0x0b, 0x0a, 0x04, 0x04, 0x00, 0x02, 0x02, 0x12, 0x03,
+    0x10, 0x02, 0x16, 0x0a, 0x0c, 0x0a, 0x05, 0x04, 0x00, 0x02, 0x02, 0x01, 0x12, 0x03, 0x10, 0x0a,
+    0x11, 0x0a, 0x0c, 0x0a, 0x05, 0x04, 0x00, 0x02, 0x02, 0x03, 0x12, 0x03, 0x10, 0x14, 0x15, 0x0a,
+    0x0a, 0x0a, 0x02, 0x04, 0x01, 0x12, 0x04, 0x13, 0x00, 0x15, 0x01, 0x0a, 0x0a, 0x0a, 0x03, 0x04,
+    0x01, 0x01, 0x12, 0x03, 0x13, 0x08, 0x1c, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+];
+include!("music.v1.serde.rs");
+include!("music.v1.tonic.rs");
+// @@protoc_insertion_point(module)

src/generated/music/v1/music.v1.serde.rs

@@ -0,0 +1,270 @@
+// @generated
+impl serde::Serialize for MonitorAlbumRequest {
+    #[allow(deprecated)]
+    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        use serde::ser::SerializeStruct;
+        let mut len = 0;
+        if !self.album.is_empty() {
+            len += 1;
+        }
+        if !self.artist.is_empty() {
+            len += 1;
+        }
+        if self.quality != 0 {
+            len += 1;
+        }
+        let mut struct_ser = serializer.serialize_struct("music.v1.MonitorAlbumRequest", len)?;
+        if !self.album.is_empty() {
+            struct_ser.serialize_field("album", &self.album)?;
+        }
+        if !self.artist.is_empty() {
+            struct_ser.serialize_field("artist", &self.artist)?;
+        }
+        if self.quality != 0 {
+            let v = Quality::try_from(self.quality).map_err(|_| {
+                serde::ser::Error::custom(format!("Invalid variant {}", self.quality))
+            })?;
+            struct_ser.serialize_field("quality", &v)?;
+        }
+        struct_ser.end()
+    }
+}
+impl<'de> serde::Deserialize<'de> for MonitorAlbumRequest {
+    #[allow(deprecated)]
+    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        const FIELDS: &[&str] = &["album", "artist", "quality"];
+
+        #[allow(clippy::enum_variant_names)]
+        enum GeneratedField {
+            Album,
+            Artist,
+            Quality,
+        }
+        impl<'de> serde::Deserialize<'de> for GeneratedField {
+            fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
+            where
+                D: serde::Deserializer<'de>,
+            {
+                struct GeneratedVisitor;
+
+                impl<'de> serde::de::Visitor<'de> for GeneratedVisitor {
+                    type Value = GeneratedField;
+
+                    fn expecting(
+                        &self,
+                        formatter: &mut std::fmt::Formatter<'_>,
+                    ) -> std::fmt::Result {
+                        write!(formatter, "expected one of: {:?}", &FIELDS)
+                    }
+
+                    #[allow(unused_variables)]
+                    fn visit_str<E>(self, value: &str) -> std::result::Result<GeneratedField, E>
+                    where
+                        E: serde::de::Error,
+                    {
+                        match value {
+                            "album" => Ok(GeneratedField::Album),
+                            "artist" => Ok(GeneratedField::Artist),
+                            "quality" => Ok(GeneratedField::Quality),
+                            _ => Err(serde::de::Error::unknown_field(value, FIELDS)),
+                        }
+                    }
+                }
+                deserializer.deserialize_identifier(GeneratedVisitor)
+            }
+        }
+        struct GeneratedVisitor;
+        impl<'de> serde::de::Visitor<'de> for GeneratedVisitor {
+            type Value = MonitorAlbumRequest;
+
+            fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+                formatter.write_str("struct music.v1.MonitorAlbumRequest")
+            }
+
+            fn visit_map<V>(self, mut map_: V) -> std::result::Result<MonitorAlbumRequest, V::Error>
+            where
+                V: serde::de::MapAccess<'de>,
+            {
+                let mut album__ = None;
+                let mut artist__ = None;
+                let mut quality__ = None;
+                while let Some(k) = map_.next_key()? {
+                    match k {
+                        GeneratedField::Album => {
+                            if album__.is_some() {
+                                return Err(serde::de::Error::duplicate_field("album"));
+                            }
+                            album__ = Some(map_.next_value()?);
+                        }
+                        GeneratedField::Artist => {
+                            if artist__.is_some() {
+                                return Err(serde::de::Error::duplicate_field("artist"));
+                            }
+                            artist__ = Some(map_.next_value()?);
+                        }
+                        GeneratedField::Quality => {
+                            if quality__.is_some() {
+                                return Err(serde::de::Error::duplicate_field("quality"));
+                            }
+                            quality__ = Some(map_.next_value::<Quality>()? as i32);
+                        }
+                    }
+                }
+                Ok(MonitorAlbumRequest {
+                    album: album__.unwrap_or_default(),
+                    artist: artist__.unwrap_or_default(),
+                    quality: quality__.unwrap_or_default(),
+                })
+            }
+        }
+        deserializer.deserialize_struct("music.v1.MonitorAlbumRequest", FIELDS, GeneratedVisitor)
+    }
+}
+impl serde::Serialize for MonitorAlbumResponse {
+    #[allow(deprecated)]
+    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        use serde::ser::SerializeStruct;
+        let len = 0;
+        let struct_ser = serializer.serialize_struct("music.v1.MonitorAlbumResponse", len)?;
+        struct_ser.end()
+    }
+}
+impl<'de> serde::Deserialize<'de> for MonitorAlbumResponse {
+    #[allow(deprecated)]
+    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        const FIELDS: &[&str] = &[];
+
+        #[allow(clippy::enum_variant_names)]
+        enum GeneratedField {}
+        impl<'de> serde::Deserialize<'de> for GeneratedField {
+            fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
+            where
+                D: serde::Deserializer<'de>,
+            {
+                struct GeneratedVisitor;
+
+                impl<'de> serde::de::Visitor<'de> for GeneratedVisitor {
+                    type Value = GeneratedField;
+
+                    fn expecting(
+                        &self,
+                        formatter: &mut std::fmt::Formatter<'_>,
+                    ) -> std::fmt::Result {
+                        write!(formatter, "expected one of: {:?}", &FIELDS)
+                    }
+
+                    #[allow(unused_variables)]
+                    fn visit_str<E>(self, value: &str) -> std::result::Result<GeneratedField, E>
+                    where
+                        E: serde::de::Error,
+                    {
+                        Err(serde::de::Error::unknown_field(value, FIELDS))
+                    }
+                }
+                deserializer.deserialize_identifier(GeneratedVisitor)
+            }
+        }
+        struct GeneratedVisitor;
+        impl<'de> serde::de::Visitor<'de> for GeneratedVisitor {
+            type Value = MonitorAlbumResponse;
+
+            fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+                formatter.write_str("struct music.v1.MonitorAlbumResponse")
+            }
+
+            fn visit_map<V>(
+                self,
+                mut map_: V,
+            ) -> std::result::Result<MonitorAlbumResponse, V::Error>
+            where
+                V: serde::de::MapAccess<'de>,
+            {
+                while map_.next_key::<GeneratedField>()?.is_some() {
+                    let _ = map_.next_value::<serde::de::IgnoredAny>()?;
+                }
+                Ok(MonitorAlbumResponse {})
+            }
+        }
+        deserializer.deserialize_struct("music.v1.MonitorAlbumResponse", FIELDS, GeneratedVisitor)
+    }
+}
+impl serde::Serialize for Quality {
+    #[allow(deprecated)]
+    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        let variant = match self {
+            Self::Mp3 => "MP3",
+            Self::Flac => "FLAC",
+        };
+        serializer.serialize_str(variant)
+    }
+}
+impl<'de> serde::Deserialize<'de> for Quality {
+    #[allow(deprecated)]
+    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        const FIELDS: &[&str] = &["MP3", "FLAC"];
+
+        struct GeneratedVisitor;
+
+        impl<'de> serde::de::Visitor<'de> for GeneratedVisitor {
+            type Value = Quality;
+
+            fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+                write!(formatter, "expected one of: {:?}", &FIELDS)
+            }
+
+            fn visit_i64<E>(self, v: i64) -> std::result::Result<Self::Value, E>
+            where
+                E: serde::de::Error,
+            {
+                i32::try_from(v)
+                    .ok()
+                    .and_then(|x| x.try_into().ok())
+                    .ok_or_else(|| {
+                        serde::de::Error::invalid_value(serde::de::Unexpected::Signed(v), &self)
+                    })
+            }
+
+            fn visit_u64<E>(self, v: u64) -> std::result::Result<Self::Value, E>
+            where
+                E: serde::de::Error,
+            {
+                i32::try_from(v)
+                    .ok()
+                    .and_then(|x| x.try_into().ok())
+                    .ok_or_else(|| {
+                        serde::de::Error::invalid_value(serde::de::Unexpected::Unsigned(v), &self)
+                    })
+            }
+
+            fn visit_str<E>(self, value: &str) -> std::result::Result<Self::Value, E>
+            where
+                E: serde::de::Error,
+            {
+                match value {
+                    "MP3" => Ok(Quality::Mp3),
+                    "FLAC" => Ok(Quality::Flac),
+                    _ => Err(serde::de::Error::unknown_variant(value, FIELDS)),
+                }
+            }
+        }
+        deserializer.deserialize_any(GeneratedVisitor)
+    }
+}

src/generated/music/v1/music.v1.tonic.rs

@@ -0,0 +1,274 @@
+// @generated
+/// Generated client implementations.
+pub mod music_service_client {
+    #![allow(
+        unused_variables,
+        dead_code,
+        missing_docs,
+        clippy::wildcard_imports,
+        clippy::let_unit_value
+    )]
+    use tonic::codegen::http::Uri;
+    use tonic::codegen::*;
+    #[derive(Debug, Clone)]
+    pub struct MusicServiceClient<T> {
+        inner: tonic::client::Grpc<T>,
+    }
+    impl MusicServiceClient<tonic::transport::Channel> {
+        /// Attempt to create a new client by connecting to a given endpoint.
+        pub async fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error>
+        where
+            D: TryInto<tonic::transport::Endpoint>,
+            D::Error: Into<StdError>,
+        {
+            let conn = tonic::transport::Endpoint::new(dst)?.connect().await?;
+            Ok(Self::new(conn))
+        }
+    }
+    impl<T> MusicServiceClient<T>
+    where
+        T: tonic::client::GrpcService<tonic::body::Body>,
+        T::Error: Into<StdError>,
+        T::ResponseBody: Body<Data = Bytes> + std::marker::Send + 'static,
+        <T::ResponseBody as Body>::Error: Into<StdError> + std::marker::Send,
+    {
+        pub fn new(inner: T) -> Self {
+            let inner = tonic::client::Grpc::new(inner);
+            Self { inner }
+        }
+        pub fn with_origin(inner: T, origin: Uri) -> Self {
+            let inner = tonic::client::Grpc::with_origin(inner, origin);
+            Self { inner }
+        }
+        pub fn with_interceptor<F>(
+            inner: T,
+            interceptor: F,
+        ) -> MusicServiceClient<InterceptedService<T, F>>
+        where
+            F: tonic::service::Interceptor,
+            T::ResponseBody: Default,
+            T: tonic::codegen::Service<
+                    http::Request<tonic::body::Body>,
+                    Response = http::Response<
+                        <T as tonic::client::GrpcService<tonic::body::Body>>::ResponseBody,
+                    >,
+                >,
+            <T as tonic::codegen::Service<http::Request<tonic::body::Body>>>::Error:
+                Into<StdError> + std::marker::Send + std::marker::Sync,
+        {
+            MusicServiceClient::new(InterceptedService::new(inner, interceptor))
+        }
+        /// Compress requests with the given encoding.
+        ///
+        /// This requires the server to support it otherwise it might respond with an
+        /// error.
+        #[must_use]
+        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
+            self.inner = self.inner.send_compressed(encoding);
+            self
+        }
+        /// Enable decompressing responses.
+        #[must_use]
+        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
+            self.inner = self.inner.accept_compressed(encoding);
+            self
+        }
+        /// Limits the maximum size of a decoded message.
+        ///
+        /// Default: `4MB`
+        #[must_use]
+        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
+            self.inner = self.inner.max_decoding_message_size(limit);
+            self
+        }
+        /// Limits the maximum size of an encoded message.
+        ///
+        /// Default: `usize::MAX`
+        #[must_use]
+        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
+            self.inner = self.inner.max_encoding_message_size(limit);
+            self
+        }
+        pub async fn monitor_album(
+            &mut self,
+            request: impl tonic::IntoRequest<super::MonitorAlbumRequest>,
+        ) -> std::result::Result<tonic::Response<super::MonitorAlbumResponse>, tonic::Status>
+        {
+            self.inner.ready().await.map_err(|e| {
+                tonic::Status::unknown(format!("Service was not ready: {}", e.into()))
+            })?;
+            let codec = tonic_prost::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static("/music.v1.MusicService/MonitorAlbum");
+            let mut req = request.into_request();
+            req.extensions_mut()
+                .insert(GrpcMethod::new("music.v1.MusicService", "MonitorAlbum"));
+            self.inner.unary(req, path, codec).await
+        }
+    }
+}
+/// Generated server implementations.
+pub mod music_service_server {
+    #![allow(
+        unused_variables,
+        dead_code,
+        missing_docs,
+        clippy::wildcard_imports,
+        clippy::let_unit_value
+    )]
+    use tonic::codegen::*;
+    /// Generated trait containing gRPC methods that should be implemented for use with MusicServiceServer.
+    #[async_trait]
+    pub trait MusicService: std::marker::Send + std::marker::Sync + 'static {
+        async fn monitor_album(
+            &self,
+            request: tonic::Request<super::MonitorAlbumRequest>,
+        ) -> std::result::Result<tonic::Response<super::MonitorAlbumResponse>, tonic::Status>;
+    }
+    #[derive(Debug)]
+    pub struct MusicServiceServer<T> {
+        inner: Arc<T>,
+        accept_compression_encodings: EnabledCompressionEncodings,
+        send_compression_encodings: EnabledCompressionEncodings,
+        max_decoding_message_size: Option<usize>,
+        max_encoding_message_size: Option<usize>,
+    }
+    impl<T> MusicServiceServer<T> {
+        pub fn new(inner: T) -> Self {
+            Self::from_arc(Arc::new(inner))
+        }
+        pub fn from_arc(inner: Arc<T>) -> Self {
+            Self {
+                inner,
+                accept_compression_encodings: Default::default(),
+                send_compression_encodings: Default::default(),
+                max_decoding_message_size: None,
+                max_encoding_message_size: None,
+            }
+        }
+        pub fn with_interceptor<F>(inner: T, interceptor: F) -> InterceptedService<Self, F>
+        where
+            F: tonic::service::Interceptor,
+        {
+            InterceptedService::new(Self::new(inner), interceptor)
+        }
+        /// Enable decompressing requests with the given encoding.
+        #[must_use]
+        pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self {
+            self.accept_compression_encodings.enable(encoding);
+            self
+        }
+        /// Compress responses with the given encoding, if the client supports it.
+        #[must_use]
+        pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self {
+            self.send_compression_encodings.enable(encoding);
+            self
+        }
+        /// Limits the maximum size of a decoded message.
+        ///
+        /// Default: `4MB`
+        #[must_use]
+        pub fn max_decoding_message_size(mut self, limit: usize) -> Self {
+            self.max_decoding_message_size = Some(limit);
+            self
+        }
+        /// Limits the maximum size of an encoded message.
+        ///
+        /// Default: `usize::MAX`
+        #[must_use]
+        pub fn max_encoding_message_size(mut self, limit: usize) -> Self {
+            self.max_encoding_message_size = Some(limit);
+            self
+        }
+    }
+    impl<T, B> tonic::codegen::Service<http::Request<B>> for MusicServiceServer<T>
+    where
+        T: MusicService,
+        B: Body + std::marker::Send + 'static,
+        B::Error: Into<StdError> + std::marker::Send + 'static,
+    {
+        type Response = http::Response<tonic::body::Body>;
+        type Error = std::convert::Infallible;
+        type Future = BoxFuture<Self::Response, Self::Error>;
+        fn poll_ready(
+            &mut self,
+            _cx: &mut Context<'_>,
+        ) -> Poll<std::result::Result<(), Self::Error>> {
+            Poll::Ready(Ok(()))
+        }
+        fn call(&mut self, req: http::Request<B>) -> Self::Future {
+            match req.uri().path() {
+                "/music.v1.MusicService/MonitorAlbum" => {
+                    #[allow(non_camel_case_types)]
+                    struct MonitorAlbumSvc<T: MusicService>(pub Arc<T>);
+                    impl<T: MusicService> tonic::server::UnaryService<super::MonitorAlbumRequest>
+                        for MonitorAlbumSvc<T>
+                    {
+                        type Response = super::MonitorAlbumResponse;
+                        type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
+                        fn call(
+                            &mut self,
+                            request: tonic::Request<super::MonitorAlbumRequest>,
+                        ) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move {
+                                <T as MusicService>::monitor_album(&inner, request).await
+                            };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = MonitorAlbumSvc(inner);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(
+                                accept_compression_encodings,
+                                send_compression_encodings,
+                            )
+                            .apply_max_message_size_config(
+                                max_decoding_message_size,
+                                max_encoding_message_size,
+                            );
+                        let res = grpc.unary(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
+                _ => Box::pin(async move {
+                    let mut response = http::Response::new(tonic::body::Body::default());
+                    let headers = response.headers_mut();
+                    headers.insert(
+                        tonic::Status::GRPC_STATUS,
+                        (tonic::Code::Unimplemented as i32).into(),
+                    );
+                    headers.insert(
+                        http::header::CONTENT_TYPE,
+                        tonic::metadata::GRPC_CONTENT_TYPE,
+                    );
+                    Ok(response)
+                }),
+            }
+        }
+    }
+    impl<T> Clone for MusicServiceServer<T> {
+        fn clone(&self) -> Self {
+            let inner = self.inner.clone();
+            Self {
+                inner,
+                accept_compression_encodings: self.accept_compression_encodings,
+                send_compression_encodings: self.send_compression_encodings,
+                max_decoding_message_size: self.max_decoding_message_size,
+                max_encoding_message_size: self.max_encoding_message_size,
+            }
+        }
+    }
+    /// Generated gRPC service name
+    pub const SERVICE_NAME: &str = "music.v1.MusicService";
+    impl<T> tonic::server::NamedService for MusicServiceServer<T> {
+        const NAME: &'static str = SERVICE_NAME;
+    }
+}

src/magnet_resolver.rs

@@ -0,0 +1,74 @@
+//! Magnet-link resolver backed by librqbit.
+//!
+//! Turns a `magnet:` URI into the same bencoded torrent bytes a `.torrent`
+//! download would yield, by resolving metadata over the BitTorrent DHT and
+//! the BEP-9 `ut_metadata` peer extension. The bytes are then handed to
+//! [`crate::torrent_parser::TorrentParser`] so magnet and torrent sources are
+//! parsed identically.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use anyhow::{Context, bail};
+use librqbit::{
+    AddTorrent, AddTorrentOptions, AddTorrentResponse, ListOnlyResponse, Session, SessionOptions,
+};
+use tempfile::TempDir;
+
+const RESOLVE_TIMEOUT: Duration = Duration::from_secs(90);
+
+pub struct MagnetResolver {
+    session: Arc<Session>,
+    _scratch_dir: TempDir,
+}
+
+impl std::fmt::Debug for MagnetResolver {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("MagnetResolver").finish_non_exhaustive()
+    }
+}
+
+impl MagnetResolver {
+    pub async fn new() -> anyhow::Result<Self> {
+        let scratch_dir = tempfile::TempDir::new()?;
+        let opts = SessionOptions {
+            disable_dht_persistence: true,
+            persistence: None,
+            enable_upnp_port_forwarding: false,
+            ..Default::default()
+        };
+        let session = Session::new_with_opts(scratch_dir.path().to_path_buf(), opts).await?;
+        Ok(Self {
+            session,
+            _scratch_dir: scratch_dir,
+        })
+    }
+
+    pub async fn resolve(&self, magnet: &str) -> anyhow::Result<Vec<u8>> {
+        let added = tokio::time::timeout(
+            RESOLVE_TIMEOUT,
+            self.session.add_torrent(
+                AddTorrent::from_url(magnet),
+                Some(AddTorrentOptions {
+                    list_only: true,
+                    ..Default::default()
+                }),
+            ),
+        )
+        .await
+        .context("magnet resolution timed out (>90s, likely no peers)")?
+        .context("error adding magnet to session")?;
+
+        match added {
+            AddTorrentResponse::ListOnly(ListOnlyResponse { torrent_bytes, .. }) => {
+                Ok(torrent_bytes.to_vec())
+            }
+            AddTorrentResponse::AlreadyManaged(_, handle) => {
+                handle.with_metadata(|m| m.torrent_bytes.to_vec())
+            }
+            AddTorrentResponse::Added(_, _) => {
+                bail!("torrent unexpectedly added under list_only")
+            }
+        }
+    }
+}

src/main.rs

@@ -1,10 +1,14 @@
 mod config;
+mod db;
 mod greeter;
 mod health;
 mod indexer;
+mod magnet_resolver;
 mod metadata;
+mod music;
 mod torrent;
 mod torrent_manager;
+mod torrent_parser;
 
 mod generated {
     pub mod hello {
@@ -23,6 +27,10 @@ mod generated {
         include!("generated/metadata/v1/metadata.v1.rs");
     }
 
+    pub mod music {
+        include!("generated/music/v1/music.v1.rs");
+    }
+
     pub mod health {
         include!("generated/health/health.rs");
     }
@@ -33,19 +41,22 @@ use std::{fs, sync::Arc};
 use generated::{
     health::health_server::HealthServer as AggregatorHealthServer,
     hello::greeter_server::GreeterServer, metadata::metadata_service_server::MetadataServiceServer,
-    torrent::torrents_server::TorrentsServer,
+    music::music_service_server::MusicServiceServer, torrent::torrents_server::TorrentsServer,
     torrent_manager::torrent_manager_server::TorrentManagerServer,
 };
 use greeter::GreeterService;
 use health::HealthService;
 use indexer::Jackett;
 use metadata::MetadataService;
+use music::MusicService;
 use tonic_health::pb::health_server::HealthServer;
 use torrent::TorrentsService;
 use torrent_manager::TorrentMananagerService;
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    tracing_subscriber::fmt().init();
+
     let config_str = fs::read_to_string("config.toml").expect("Failed to read file");
     let config: config::Config = toml::from_str(&config_str).expect("Failed to parse TOML");
     let addr = config.service.address.parse()?;
@@ -56,6 +67,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .register_encoded_file_descriptor_set(generated::health::FILE_DESCRIPTOR_SET)
         .register_encoded_file_descriptor_set(generated::torrent_manager::FILE_DESCRIPTOR_SET)
         .register_encoded_file_descriptor_set(generated::metadata::FILE_DESCRIPTOR_SET)
+        .register_encoded_file_descriptor_set(generated::music::FILE_DESCRIPTOR_SET)
         .register_encoded_file_descriptor_set(tonic_health::pb::FILE_DESCRIPTOR_SET)
         .build_v1()?;
 
@@ -65,6 +77,21 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let indexer = Arc::new(Jackett::new(&config.indexer));
     let torrent_manager = TorrentMananagerService::new(torrents.clone(), indexer);
 
+    let db_pool = match db::connect(&config.database.url).await {
+        Ok(pool) => {
+            if let Err(e) = db::apply_schema(&pool).await {
+                tracing::warn!(error = %e, "failed to apply DB schema");
+            }
+            Some(pool)
+        }
+        Err(e) => {
+            tracing::warn!(error = %e, "failed to connect to database; caching disabled");
+            None
+        }
+    };
+
+    let music = MusicService::new(metadata.clone(), torrent_manager.clone(), db_pool);
+
     tonic::transport::Server::builder()
         .add_service(reflection)
         .add_service(HealthServer::new(health.clone()))
@@ -73,6 +100,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .add_service(TorrentsServer::new(torrents))
         .add_service(TorrentManagerServer::new(torrent_manager))
         .add_service(MetadataServiceServer::new(metadata))
+        .add_service(MusicServiceServer::new(music))
         .serve(addr)
         .await?;
 

src/music.rs

@@ -0,0 +1,95 @@
+use tonic::{Request, Response, Status};
+
+use crate::{
+    generated::{
+        metadata::{
+            Album, SearchAlbumsRequest, metadata_service_server::MetadataService as MetadataSvc,
+        },
+        music::{
+            MonitorAlbumRequest, MonitorAlbumResponse, music_service_server::MusicService as Music,
+        },
+        torrent_manager::{SearchRequest, torrent_manager_server::TorrentManager},
+    },
+    magnet_resolver::MagnetResolver,
+    metadata::MetadataService,
+    torrent_manager::TorrentMananagerService,
+    torrent_parser::{ParsedTorrent, TorrentParser},
+};
+
+pub struct MusicService {
+    metadata_service: MetadataService,
+    torrent_manager: TorrentMananagerService,
+    db: Option<sqlx::PgPool>,
+}
+
+impl MusicService {
+    pub fn new(
+        metadata_service: MetadataService,
+        torrent_manager: TorrentMananagerService,
+        db: Option<sqlx::PgPool>,
+    ) -> Self {
+        Self {
+            metadata_service,
+            torrent_manager,
+            db,
+        }
+    }
+}
+
+#[tonic::async_trait]
+impl Music for MusicService {
+    async fn monitor_album(
+        &self,
+        request: Request<MonitorAlbumRequest>,
+    ) -> Result<Response<MonitorAlbumResponse>, Status> {
+        let req: &MonitorAlbumRequest = request.get_ref();
+
+        let search_request = SearchAlbumsRequest {
+            query: req.album.clone(),
+            artist: req.artist.clone(),
+            ..Default::default()
+        };
+
+        let search_response = self
+            .metadata_service
+            .search_albums(Request::new(search_request))
+            .await?;
+
+        dbg!(&search_response);
+
+        let album: &Album = &search_response.get_ref().albums[0];
+        let name = format!(
+            "{} {}",
+            album.artists[0].artist.clone().unwrap().name,
+            album.title
+        );
+
+        let torrent_search_request = SearchRequest { name: name };
+
+        dbg!(&torrent_search_request);
+        let torrent_search_response = self
+            .torrent_manager
+            .search(Request::new(torrent_search_request))
+            .await?;
+        dbg!(&torrent_search_response);
+
+        let resolver = match MagnetResolver::new().await {
+            Ok(r) => Some(std::sync::Arc::new(r)),
+            Err(e) => {
+                tracing::warn!(
+                    error = %e,
+                    "failed to init MagnetResolver; magnet links will be info-hash only"
+                );
+                None
+            }
+        };
+        let parser =
+            std::sync::Arc::new(TorrentParser::new_with_resolver(resolver, self.db.clone()));
+        let parsed_torrents: Vec<ParsedTorrent> = parser
+            .parse_search_response(torrent_search_response.get_ref())
+            .await;
+        dbg!(&parsed_torrents);
+
+        Ok(Response::new(MonitorAlbumResponse::default()))
+    }
+}

src/torrent_manager.rs

@@ -11,6 +11,7 @@ use crate::{
     torrent::TorrentsService,
 };
 
+#[derive(Clone)]
 pub struct TorrentMananagerService {
     torrents_service: TorrentsService,
     indexer: Arc<dyn Indexer>,