From c7252a12faf2479aa4ccf237f5bbe92dc4fda5b6 Mon Sep 17 00:00:00 2001 From: Tsiry Sandratraina Date: Sun, 7 Sep 2025 20:10:32 +0300 Subject: [PATCH] [tracklist] implement track queue service --- .github/workflows/ci.yml | 13 +- Cargo.lock | 551 ++++++----- apps/api/lexicons/player/addItemsToQueue.json | 4 + apps/api/lexicons/player/playDirectory.json | 9 + apps/api/pkl/defs/player/addItemsToQueue.pkl | 4 + apps/api/pkl/defs/player/playDirectory.pkl | 9 + apps/api/src/lexicon/lexicons.ts | 13 + .../app/rocksky/player/addItemsToQueue.ts | 2 + .../types/app/rocksky/player/playDirectory.ts | 3 + crates/analytics/Cargo.toml | 2 +- crates/analytics/src/handlers/albums.rs | 4 +- crates/analytics/src/handlers/artists.rs | 4 +- crates/analytics/src/handlers/scrobbles.rs | 2 +- crates/analytics/src/handlers/stats.rs | 4 +- crates/analytics/src/handlers/tracks.rs | 2 +- crates/analytics/src/lib.rs | 42 + crates/connect/Cargo.toml | 2 +- crates/dropbox/Cargo.toml | 2 +- crates/dropbox/src/lib.rs | 10 + crates/googledrive/Cargo.toml | 2 +- crates/googledrive/src/lib.rs | 10 + crates/jetstream/Cargo.toml | 2 +- crates/jetstream/src/lib.rs | 24 + crates/jetstream/src/main.rs | 2 +- crates/playlists/Cargo.toml | 2 +- crates/playlists/src/core.rs | 2 +- crates/playlists/src/lib.rs | 59 ++ crates/playlists/src/main.rs | 2 +- crates/rockskyd/Cargo.toml | 22 + crates/rockskyd/src/cmd/analytics.rs | 11 + crates/rockskyd/src/cmd/dropbox.rs | 11 + crates/rockskyd/src/cmd/googledrive.rs | 11 + crates/rockskyd/src/cmd/jetstream.rs | 6 + crates/rockskyd/src/cmd/mod.rs | 9 + crates/rockskyd/src/cmd/playlist.rs | 6 + crates/rockskyd/src/cmd/scrobbler.rs | 6 + crates/rockskyd/src/cmd/spotify.rs | 6 + crates/rockskyd/src/cmd/tracklist.rs | 6 + crates/rockskyd/src/cmd/webscrobbler.rs | 6 + crates/rockskyd/src/main.rs | 82 ++ crates/scrobbler/Cargo.toml | 2 +- crates/scrobbler/src/lib.rs | 103 ++ crates/spotify/Cargo.toml | 2 +- crates/spotify/src/lib.rs | 897 ++++++++++++++++++ crates/spotify/src/main.rs | 691 +------------- crates/storage/Cargo.toml | 2 +- crates/storage/src/lib.rs | 1 + crates/tracklist/Cargo.toml | 24 + crates/tracklist/src/handlers/mod.rs | 47 + crates/tracklist/src/handlers/tracklist.rs | 205 ++++ crates/tracklist/src/lib.rs | 11 + crates/tracklist/src/queue.rs | 609 ++++++++++++ crates/tracklist/src/server.rs | 60 ++ crates/tracklist/src/types.rs | 87 ++ crates/webscrobbler/Cargo.toml | 2 +- crates/webscrobbler/src/consts.rs | 10 + crates/webscrobbler/src/handlers.rs | 2 +- crates/webscrobbler/src/lib.rs | 78 ++ crates/webscrobbler/src/main.rs | 84 +- 59 files changed, 2841 insertions(+), 1045 deletions(-) create mode 100644 crates/dropbox/src/lib.rs create mode 100644 crates/googledrive/src/lib.rs create mode 100644 crates/jetstream/src/lib.rs create mode 100644 crates/rockskyd/Cargo.toml create mode 100644 crates/rockskyd/src/cmd/analytics.rs create mode 100644 crates/rockskyd/src/cmd/dropbox.rs create mode 100644 crates/rockskyd/src/cmd/googledrive.rs create mode 100644 crates/rockskyd/src/cmd/jetstream.rs create mode 100644 crates/rockskyd/src/cmd/mod.rs create mode 100644 crates/rockskyd/src/cmd/playlist.rs create mode 100644 crates/rockskyd/src/cmd/scrobbler.rs create mode 100644 crates/rockskyd/src/cmd/spotify.rs create mode 100644 crates/rockskyd/src/cmd/tracklist.rs create mode 100644 crates/rockskyd/src/cmd/webscrobbler.rs create mode 100644 crates/rockskyd/src/main.rs create mode 100644 crates/scrobbler/src/lib.rs create mode 100644 crates/spotify/src/lib.rs create mode 100644 crates/storage/src/lib.rs create mode 100644 crates/tracklist/Cargo.toml create mode 100644 crates/tracklist/src/handlers/mod.rs create mode 100644 crates/tracklist/src/handlers/tracklist.rs create mode 100644 crates/tracklist/src/lib.rs create mode 100644 crates/tracklist/src/queue.rs create mode 100644 crates/tracklist/src/server.rs create mode 100644 crates/tracklist/src/types.rs create mode 100644 crates/webscrobbler/src/consts.rs create mode 100644 crates/webscrobbler/src/lib.rs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bc5ec2a5..b16ce5de 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -7,10 +7,18 @@ on: branches: - main jobs: - fmt: + tests: runs-on: ubuntu-latest steps: - uses: actions/checkout@v2 + - name: Install dependencies + run: | + apt-get update && apt-get install -y \ + libreadline-dev \ + pkg-config \ + flex \ + bison \ + build-essential - name: Setup Fluent CI uses: fluentci-io/setup-fluentci@v5 with: @@ -23,3 +31,6 @@ jobs: run: | type cargo cargo fmt --all --check + - name: Run tests + run: | + cargo test -p rocksky-tracklist diff --git a/Cargo.lock b/Cargo.lock index de42ee17..ac1f99ad 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -49,7 +49,7 @@ dependencies = [ "mime", "percent-encoding", "pin-project-lite", - "rand 0.9.1", + "rand 0.9.2", "sha1", "smallvec", "tokio", @@ -340,26 +340,6 @@ version = "0.2.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" -[[package]] -name = "analytics" -version = "0.1.0" -dependencies = [ - "actix-web", - "anyhow", - "async-nats", - "chrono", - "clap", - "dotenv", - "duckdb", - "owo-colors", - "polars", - "serde", - "serde_json", - "sqlx", - "tokio", - "tokio-stream", -] - [[package]] name = "android-tzdata" version = "0.1.1" @@ -1275,27 +1255,6 @@ dependencies = [ "crossbeam-utils", ] -[[package]] -name = "connect" -version = "0.1.0" -dependencies = [ - "anyhow", - "async-trait", - "base64 0.22.1", - "dirs", - "futures-util", - "http 1.3.1", - "jsonrpsee", - "owo-colors", - "reqwest", - "serde", - "serde_json", - "tokio", - "tokio-stream", - "tokio-tungstenite", - "tungstenite", -] - [[package]] name = "const-oid" version = "0.9.6" @@ -1678,36 +1637,6 @@ version = "0.15.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b" -[[package]] -name = "dropbox" -version = "0.1.0" -dependencies = [ - "actix-web", - "aes", - "anyhow", - "async-nats", - "chrono", - "clap", - "ctr", - "dotenv", - "futures", - "hex", - "jsonwebtoken", - "lofty", - "md5", - "owo-colors", - "redis 0.29.5", - "reqwest", - "serde", - "serde_json", - "sha256", - "sqlx", - "symphonia", - "tempfile", - "tokio", - "tokio-stream", -] - [[package]] name = "duckdb" version = "1.2.2" @@ -2183,37 +2112,6 @@ dependencies = [ "web-sys", ] -[[package]] -name = "googledrive" -version = "0.1.0" -dependencies = [ - "actix-web", - "aes", - "anyhow", - "async-nats", - "chrono", - "clap", - "ctr", - "dotenv", - "futures", - "hex", - "jsonwebtoken", - "lofty", - "md5", - "owo-colors", - "redis 0.29.5", - "reqwest", - "serde", - "serde_json", - "serde_urlencoded", - "sha256", - "sqlx", - "symphonia", - "tempfile", - "tokio", - "tokio-stream", -] - [[package]] name = "h2" version = "0.3.26" @@ -2772,28 +2670,6 @@ version = "1.0.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4a5f13b858c8d314ee3e8f639011f7ccefe71f97f96e50151fb991f267928e2c" -[[package]] -name = "jetstream" -version = "0.1.0" -dependencies = [ - "anyhow", - "async-nats", - "chrono", - "dotenv", - "futures-util", - "owo-colors", - "reqwest", - "serde", - "serde_json", - "sha256", - "sqlx", - "tokio", - "tokio-stream", - "tokio-tungstenite", - "tungstenite", - "url", -] - [[package]] name = "jni" version = "0.21.1" @@ -3748,7 +3624,7 @@ dependencies = [ "pgrx-pg-config", "postgres", "proptest", - "rand 0.9.1", + "rand 0.9.2", "regex", "serde", "serde_json", @@ -3865,31 +3741,6 @@ dependencies = [ "array-init-cursor", ] -[[package]] -name = "playlists" -version = "0.1.0" -dependencies = [ - "aes", - "anyhow", - "async-nats", - "chrono", - "clap", - "ctr", - "dotenv", - "duckdb", - "hex", - "jsonwebtoken", - "owo-colors", - "polars", - "reqwest", - "serde", - "serde_json", - "sha2", - "sqlx", - "tokio", - "tokio-stream", -] - [[package]] name = "polars" version = "0.46.0" @@ -4421,7 +4272,7 @@ dependencies = [ "hmac", "md-5", "memchr", - "rand 0.9.1", + "rand 0.9.2", "sha2", "stringprep", ] @@ -4490,7 +4341,7 @@ dependencies = [ "bitflags 2.9.1", "lazy_static", "num-traits", - "rand 0.9.1", + "rand 0.9.2", "rand_chacha 0.9.0", "rand_xorshift", "regex-syntax", @@ -4583,7 +4434,7 @@ dependencies = [ "bytes", "getrandom 0.3.3", "lru-slab", - "rand 0.9.1", + "rand 0.9.2", "ring", "rustc-hash", "rustls 0.23.27", @@ -4657,9 +4508,9 @@ dependencies = [ [[package]] name = "rand" -version = "0.9.1" +version = "0.9.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9fbfd9d094a40bf3ae768db9361049ace4c0e04a4fd6b359518bd7b73a73dd97" +checksum = "6db2770f06117d490610c7488547d543617b21bfa07796d7a12f6f1bd53850d1" dependencies = [ "rand_chacha 0.9.0", "rand_core 0.9.3", @@ -4797,13 +4648,21 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1bc42f3a12fd4408ce64d8efef67048a924e543bd35c6591c0447fda9054695f" dependencies = [ "arc-swap", + "bytes", "combine", + "futures-util", "itoa", "num-bigint", "percent-encoding", + "pin-project-lite", + "rustls 0.23.27", + "rustls-native-certs 0.8.1", "ryu", "sha1_smol", "socket2", + "tokio", + "tokio-rustls 0.26.2", + "tokio-util", "url", ] @@ -4960,6 +4819,286 @@ dependencies = [ "syn 1.0.109", ] +[[package]] +name = "rocksky-analytics" +version = "0.1.0" +dependencies = [ + "actix-web", + "anyhow", + "async-nats", + "chrono", + "clap", + "dotenv", + "duckdb", + "owo-colors", + "polars", + "serde", + "serde_json", + "sqlx", + "tokio", + "tokio-stream", +] + +[[package]] +name = "rocksky-connect" +version = "0.1.0" +dependencies = [ + "anyhow", + "async-trait", + "base64 0.22.1", + "dirs", + "futures-util", + "http 1.3.1", + "jsonrpsee", + "owo-colors", + "reqwest", + "serde", + "serde_json", + "tokio", + "tokio-stream", + "tokio-tungstenite", + "tungstenite", +] + +[[package]] +name = "rocksky-dropbox" +version = "0.1.0" +dependencies = [ + "actix-web", + "aes", + "anyhow", + "async-nats", + "chrono", + "clap", + "ctr", + "dotenv", + "futures", + "hex", + "jsonwebtoken", + "lofty", + "md5", + "owo-colors", + "redis 0.29.5", + "reqwest", + "serde", + "serde_json", + "sha256", + "sqlx", + "symphonia", + "tempfile", + "tokio", + "tokio-stream", +] + +[[package]] +name = "rocksky-googledrive" +version = "0.1.0" +dependencies = [ + "actix-web", + "aes", + "anyhow", + "async-nats", + "chrono", + "clap", + "ctr", + "dotenv", + "futures", + "hex", + "jsonwebtoken", + "lofty", + "md5", + "owo-colors", + "redis 0.29.5", + "reqwest", + "serde", + "serde_json", + "serde_urlencoded", + "sha256", + "sqlx", + "symphonia", + "tempfile", + "tokio", + "tokio-stream", +] + +[[package]] +name = "rocksky-jetstream" +version = "0.1.0" +dependencies = [ + "anyhow", + "async-nats", + "chrono", + "dotenv", + "futures-util", + "owo-colors", + "reqwest", + "serde", + "serde_json", + "sha256", + "sqlx", + "tokio", + "tokio-stream", + "tokio-tungstenite", + "tungstenite", + "url", +] + +[[package]] +name = "rocksky-playlists" +version = "0.1.0" +dependencies = [ + "aes", + "anyhow", + "async-nats", + "chrono", + "clap", + "ctr", + "dotenv", + "duckdb", + "hex", + "jsonwebtoken", + "owo-colors", + "polars", + "reqwest", + "serde", + "serde_json", + "sha2", + "sqlx", + "tokio", + "tokio-stream", +] + +[[package]] +name = "rocksky-scrobbler" +version = "0.1.0" +dependencies = [ + "actix-limitation", + "actix-session 0.10.1", + "actix-web", + "aes", + "anyhow", + "chrono", + "ctr", + "dotenv", + "hex", + "jsonwebtoken", + "md5", + "owo-colors", + "quick-xml 0.37.5", + "rand 0.9.2", + "redis 0.29.5", + "reqwest", + "serde", + "serde_json", + "sqlx", + "tokio", + "tokio-stream", + "uuid", +] + +[[package]] +name = "rocksky-spotify" +version = "0.1.0" +dependencies = [ + "aes", + "anyhow", + "async-nats", + "chrono", + "ctr", + "dotenv", + "hex", + "jsonwebtoken", + "owo-colors", + "redis 0.29.5", + "reqwest", + "serde", + "serde_json", + "sqlx", + "tokio", + "tokio-stream", +] + +[[package]] +name = "rocksky-storage" +version = "0.1.0" +dependencies = [ + "actix-web", + "anyhow", + "dotenv", + "owo-colors", + "rust-s3", + "serde", + "serde_json", + "sqlx", + "tokio", + "tokio-stream", +] + +[[package]] +name = "rocksky-tracklist" +version = "0.1.0" +dependencies = [ + "actix-web", + "anyhow", + "async-nats", + "clap", + "dotenv", + "owo-colors", + "polars", + "rand 0.9.2", + "redis 0.29.5", + "serde", + "serde_json", + "tokio", + "tokio-stream", + "uuid", +] + +[[package]] +name = "rocksky-webscrobbler" +version = "0.1.0" +dependencies = [ + "actix-limitation", + "actix-session 0.10.1", + "actix-web", + "aes", + "anyhow", + "chrono", + "ctr", + "dotenv", + "hex", + "jsonwebtoken", + "md5", + "owo-colors", + "rand 0.9.2", + "redis 0.29.5", + "reqwest", + "serde", + "serde_json", + "sqlx", + "tokio", + "tokio-stream", +] + +[[package]] +name = "rockskyd" +version = "0.1.0" +dependencies = [ + "anyhow", + "clap", + "dotenv", + "rocksky-analytics", + "rocksky-dropbox", + "rocksky-googledrive", + "rocksky-jetstream", + "rocksky-playlists", + "rocksky-scrobbler", + "rocksky-spotify", + "rocksky-tracklist", + "rocksky-webscrobbler", + "tokio", +] + [[package]] name = "rsa" version = "0.9.8" @@ -5289,34 +5428,6 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" -[[package]] -name = "scrobbler" -version = "0.1.0" -dependencies = [ - "actix-limitation", - "actix-session 0.10.1", - "actix-web", - "aes", - "anyhow", - "chrono", - "ctr", - "dotenv", - "hex", - "jsonwebtoken", - "md5", - "owo-colors", - "quick-xml 0.37.5", - "rand 0.9.1", - "redis 0.29.5", - "reqwest", - "serde", - "serde_json", - "sqlx", - "tokio", - "tokio-stream", - "uuid", -] - [[package]] name = "sct" version = "0.7.1" @@ -5649,28 +5760,6 @@ dependencies = [ "der", ] -[[package]] -name = "spotify" -version = "0.1.0" -dependencies = [ - "aes", - "anyhow", - "async-nats", - "chrono", - "ctr", - "dotenv", - "hex", - "jsonwebtoken", - "owo-colors", - "redis 0.29.5", - "reqwest", - "serde", - "serde_json", - "sqlx", - "tokio", - "tokio-stream", -] - [[package]] name = "sptr" version = "0.3.2" @@ -5905,22 +5994,6 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" -[[package]] -name = "storage" -version = "0.1.0" -dependencies = [ - "actix-web", - "anyhow", - "dotenv", - "owo-colors", - "rust-s3", - "serde", - "serde_json", - "sqlx", - "tokio", - "tokio-stream", -] - [[package]] name = "streaming-decompression" version = "0.1.2" @@ -6478,7 +6551,7 @@ dependencies = [ "pin-project-lite", "postgres-protocol", "postgres-types", - "rand 0.9.1", + "rand 0.9.2", "socket2", "tokio", "tokio-util", @@ -6741,7 +6814,7 @@ dependencies = [ "http 1.3.1", "httparse", "log", - "rand 0.9.1", + "rand 0.9.2", "rustls 0.23.27", "rustls-pki-types", "sha1", @@ -6874,9 +6947,9 @@ checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" [[package]] name = "uuid" -version = "1.17.0" +version = "1.18.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3cf4199d1e5d15ddd86a694e4d0dffa9c323ce759fea589f00fef9d81cc1931d" +checksum = "f33196643e165781c20a5ead5582283a7dacbb87855d867fbc2df3f81eddc1be" dependencies = [ "getrandom 0.3.3", "js-sys", @@ -7075,32 +7148,6 @@ dependencies = [ "rustls-pki-types", ] -[[package]] -name = "webscrobbler" -version = "0.1.0" -dependencies = [ - "actix-limitation", - "actix-session 0.10.1", - "actix-web", - "aes", - "anyhow", - "chrono", - "ctr", - "dotenv", - "hex", - "jsonwebtoken", - "md5", - "owo-colors", - "rand 0.9.1", - "redis 0.29.5", - "reqwest", - "serde", - "serde_json", - "sqlx", - "tokio", - "tokio-stream", -] - [[package]] name = "whoami" version = "1.6.0" diff --git a/apps/api/lexicons/player/addItemsToQueue.json b/apps/api/lexicons/player/addItemsToQueue.json index 4e75805a..811498e3 100644 --- a/apps/api/lexicons/player/addItemsToQueue.json +++ b/apps/api/lexicons/player/addItemsToQueue.json @@ -24,6 +24,10 @@ "position": { "type": "integer", "description": "Position in the queue to insert the items at, defaults to the end if not specified" + }, + "shuffle": { + "type": "boolean", + "description": "Whether to shuffle the added items in the queue" } } } diff --git a/apps/api/lexicons/player/playDirectory.json b/apps/api/lexicons/player/playDirectory.json index a9a5985b..2aa66097 100644 --- a/apps/api/lexicons/player/playDirectory.json +++ b/apps/api/lexicons/player/playDirectory.json @@ -16,6 +16,15 @@ }, "directoryId": { "type": "string" + }, + "shuffle": { + "type": "boolean" + }, + "recurse": { + "type": "boolean" + }, + "position": { + "type": "integer" } } } diff --git a/apps/api/pkl/defs/player/addItemsToQueue.pkl b/apps/api/pkl/defs/player/addItemsToQueue.pkl index 25b6a88e..6e4f8f97 100644 --- a/apps/api/pkl/defs/player/addItemsToQueue.pkl +++ b/apps/api/pkl/defs/player/addItemsToQueue.pkl @@ -24,6 +24,10 @@ defs = new Mapping { type = "integer" description = "Position in the queue to insert the items at, defaults to the end if not specified" } + ["shuffle"] = new BooleanType { + type = "boolean" + description = "Whether to shuffle the added items in the queue" + } } } } diff --git a/apps/api/pkl/defs/player/playDirectory.pkl b/apps/api/pkl/defs/player/playDirectory.pkl index 46758535..4b0dbf31 100644 --- a/apps/api/pkl/defs/player/playDirectory.pkl +++ b/apps/api/pkl/defs/player/playDirectory.pkl @@ -16,6 +16,15 @@ defs = new Mapping { ["directoryId"] = new StringType { type = "string" } + ["shuffle"] = new BooleanType { + type = "boolean" + } + ["recurse"] = new BooleanType { + type = "boolean" + } + ["position"] = new IntegerType { + type = "integer" + } } } } diff --git a/apps/api/src/lexicon/lexicons.ts b/apps/api/src/lexicon/lexicons.ts index 1bd69447..34122fdd 100644 --- a/apps/api/src/lexicon/lexicons.ts +++ b/apps/api/src/lexicon/lexicons.ts @@ -1892,6 +1892,10 @@ export const schemaDict = { description: 'Position in the queue to insert the items at, defaults to the end if not specified', }, + shuffle: { + type: 'boolean', + description: 'Whether to shuffle the added items in the queue', + }, }, }, }, @@ -2033,6 +2037,15 @@ export const schemaDict = { directoryId: { type: 'string', }, + shuffle: { + type: 'boolean', + }, + recurse: { + type: 'boolean', + }, + position: { + type: 'integer', + }, }, }, }, diff --git a/apps/api/src/lexicon/types/app/rocksky/player/addItemsToQueue.ts b/apps/api/src/lexicon/types/app/rocksky/player/addItemsToQueue.ts index 11482efd..9388d3ab 100644 --- a/apps/api/src/lexicon/types/app/rocksky/player/addItemsToQueue.ts +++ b/apps/api/src/lexicon/types/app/rocksky/player/addItemsToQueue.ts @@ -13,6 +13,8 @@ export interface QueryParams { items: string[] /** Position in the queue to insert the items at, defaults to the end if not specified */ position?: number + /** Whether to shuffle the added items in the queue */ + shuffle?: boolean } export type InputSchema = undefined diff --git a/apps/api/src/lexicon/types/app/rocksky/player/playDirectory.ts b/apps/api/src/lexicon/types/app/rocksky/player/playDirectory.ts index bf6cef02..1ebbc34e 100644 --- a/apps/api/src/lexicon/types/app/rocksky/player/playDirectory.ts +++ b/apps/api/src/lexicon/types/app/rocksky/player/playDirectory.ts @@ -11,6 +11,9 @@ import { HandlerAuth, HandlerPipeThrough } from '@atproto/xrpc-server' export interface QueryParams { playerId?: string directoryId: string + shuffle?: boolean + recurse?: boolean + position?: number } export type InputSchema = undefined diff --git a/crates/analytics/Cargo.toml b/crates/analytics/Cargo.toml index bb2a11be..b014274b 100644 --- a/crates/analytics/Cargo.toml +++ b/crates/analytics/Cargo.toml @@ -1,5 +1,5 @@ [package] -name = "analytics" +name = "rocksky-analytics" version = "0.1.0" authors.workspace = true edition.workspace = true diff --git a/crates/analytics/src/handlers/albums.rs b/crates/analytics/src/handlers/albums.rs index 72abc507..6a71b9c3 100644 --- a/crates/analytics/src/handlers/albums.rs +++ b/crates/analytics/src/handlers/albums.rs @@ -1,10 +1,10 @@ use std::sync::{Arc, Mutex}; -use actix_web::{web, HttpRequest, HttpResponse}; -use analytics::types::{ +use crate::types::{ album::{Album, GetAlbumTracksParams, GetAlbumsParams, GetTopAlbumsParams}, track::Track, }; +use actix_web::{web, HttpRequest, HttpResponse}; use anyhow::Error; use duckdb::Connection; use tokio_stream::StreamExt; diff --git a/crates/analytics/src/handlers/artists.rs b/crates/analytics/src/handlers/artists.rs index 4b26ad1d..f4a5dd8d 100644 --- a/crates/analytics/src/handlers/artists.rs +++ b/crates/analytics/src/handlers/artists.rs @@ -1,13 +1,13 @@ use std::sync::{Arc, Mutex}; -use actix_web::{web, HttpRequest, HttpResponse}; -use analytics::types::{ +use crate::types::{ album::Album, artist::{ Artist, GetArtistAlbumsParams, GetArtistTracksParams, GetArtistsParams, GetTopArtistsParams, }, track::Track, }; +use actix_web::{web, HttpRequest, HttpResponse}; use anyhow::Error; use duckdb::Connection; use tokio_stream::StreamExt; diff --git a/crates/analytics/src/handlers/scrobbles.rs b/crates/analytics/src/handlers/scrobbles.rs index 17d41d81..b752dd1b 100644 --- a/crates/analytics/src/handlers/scrobbles.rs +++ b/crates/analytics/src/handlers/scrobbles.rs @@ -1,7 +1,7 @@ use std::sync::{Arc, Mutex}; +use crate::types::scrobble::{GetScrobblesParams, ScrobbleTrack}; use actix_web::{web, HttpRequest, HttpResponse}; -use analytics::types::scrobble::{GetScrobblesParams, ScrobbleTrack}; use anyhow::Error; use duckdb::Connection; use tokio_stream::StreamExt; diff --git a/crates/analytics/src/handlers/stats.rs b/crates/analytics/src/handlers/stats.rs index 850fb256..18c4153c 100644 --- a/crates/analytics/src/handlers/stats.rs +++ b/crates/analytics/src/handlers/stats.rs @@ -1,8 +1,7 @@ use std::sync::{Arc, Mutex}; use crate::read_payload; -use actix_web::{web, HttpRequest, HttpResponse}; -use analytics::types::{ +use crate::types::{ scrobble::{ScrobblesPerDay, ScrobblesPerMonth, ScrobblesPerYear}, stats::{ GetAlbumScrobblesParams, GetArtistScrobblesParams, GetScrobblesPerDayParams, @@ -10,6 +9,7 @@ use analytics::types::{ GetTrackScrobblesParams, }, }; +use actix_web::{web, HttpRequest, HttpResponse}; use anyhow::Error; use duckdb::Connection; use serde_json::json; diff --git a/crates/analytics/src/handlers/tracks.rs b/crates/analytics/src/handlers/tracks.rs index 4524925f..761dbc8c 100644 --- a/crates/analytics/src/handlers/tracks.rs +++ b/crates/analytics/src/handlers/tracks.rs @@ -1,7 +1,7 @@ use std::sync::{Arc, Mutex}; +use crate::types::track::{GetLovedTracksParams, GetTopTracksParams, GetTracksParams, Track}; use actix_web::{web, HttpRequest, HttpResponse}; -use analytics::types::track::{GetLovedTracksParams, GetTopTracksParams, GetTracksParams, Track}; use anyhow::Error; use duckdb::Connection; use tokio_stream::StreamExt; diff --git a/crates/analytics/src/lib.rs b/crates/analytics/src/lib.rs index 20cf7aa5..554f9d72 100644 --- a/crates/analytics/src/lib.rs +++ b/crates/analytics/src/lib.rs @@ -1,2 +1,44 @@ +use std::{ + env, + sync::{Arc, Mutex}, +}; + +use anyhow::Error; +use duckdb::Connection; +use sqlx::postgres::PgPoolOptions; + +use crate::core::create_tables; + +pub mod cmd; +pub mod core; +pub mod handlers; +pub mod subscriber; pub mod types; pub mod xata; + +pub async fn serve() -> Result<(), Error> { + let conn = Connection::open("./rocksky-analytics.ddb")?; + + create_tables(&conn).await?; + + let conn = Arc::new(Mutex::new(conn)); + cmd::serve::serve(conn).await?; + + Ok(()) +} + +pub async fn sync() -> Result<(), Error> { + let pool = PgPoolOptions::new() + .max_connections(5) + .connect(&env::var("XATA_POSTGRES_URL")?) + .await?; + + let conn = Connection::open("./rocksky-analytics.ddb")?; + create_tables(&conn).await?; + + let conn = Arc::new(Mutex::new(conn)); + + cmd::sync::sync(conn, &pool).await?; + + Ok(()) +} diff --git a/crates/connect/Cargo.toml b/crates/connect/Cargo.toml index 51c4b7a7..2f113a15 100644 --- a/crates/connect/Cargo.toml +++ b/crates/connect/Cargo.toml @@ -1,5 +1,5 @@ [package] -name = "connect" +name = "rocksky-connect" version = "0.1.0" authors = ["Tsiry Sandratraina "] edition = "2024" diff --git a/crates/dropbox/Cargo.toml b/crates/dropbox/Cargo.toml index 080bd227..ceddae8c 100644 --- a/crates/dropbox/Cargo.toml +++ b/crates/dropbox/Cargo.toml @@ -1,5 +1,5 @@ [package] -name = "dropbox" +name = "rocksky-dropbox" version = "0.1.0" authors.workspace = true edition.workspace = true diff --git a/crates/dropbox/src/lib.rs b/crates/dropbox/src/lib.rs new file mode 100644 index 00000000..c1b115bd --- /dev/null +++ b/crates/dropbox/src/lib.rs @@ -0,0 +1,10 @@ +pub mod client; +pub mod cmd; +pub mod consts; +pub mod crypto; +pub mod handlers; +pub mod repo; +pub mod scan; +pub mod token; +pub mod types; +pub mod xata; diff --git a/crates/googledrive/Cargo.toml b/crates/googledrive/Cargo.toml index f0069a1a..464f26be 100644 --- a/crates/googledrive/Cargo.toml +++ b/crates/googledrive/Cargo.toml @@ -1,5 +1,5 @@ [package] -name = "googledrive" +name = "rocksky-googledrive" version = "0.1.0" authors.workspace = true edition.workspace = true diff --git a/crates/googledrive/src/lib.rs b/crates/googledrive/src/lib.rs new file mode 100644 index 00000000..c1b115bd --- /dev/null +++ b/crates/googledrive/src/lib.rs @@ -0,0 +1,10 @@ +pub mod client; +pub mod cmd; +pub mod consts; +pub mod crypto; +pub mod handlers; +pub mod repo; +pub mod scan; +pub mod token; +pub mod types; +pub mod xata; diff --git a/crates/jetstream/Cargo.toml b/crates/jetstream/Cargo.toml index d6921235..3c74a4e7 100644 --- a/crates/jetstream/Cargo.toml +++ b/crates/jetstream/Cargo.toml @@ -1,5 +1,5 @@ [package] -name = "jetstream" +name = "rocksky-jetstream" version = "0.1.0" authors.workspace = true edition.workspace = true diff --git a/crates/jetstream/src/lib.rs b/crates/jetstream/src/lib.rs new file mode 100644 index 00000000..56e29d41 --- /dev/null +++ b/crates/jetstream/src/lib.rs @@ -0,0 +1,24 @@ +use anyhow::Error; +use std::env; + +use subscriber::ScrobbleSubscriber; + +pub mod profile; +pub mod repo; +pub mod subscriber; +pub mod types; +pub mod xata; + +pub async fn subscribe() -> Result<(), Error> { + let jetstream_server = env::var("JETSTREAM_SERVER") + .unwrap_or_else(|_| "wss://jetstream2.us-west.bsky.network".to_string()); + let url = format!( + "{}/subscribe?wantedCollections=app.rocksky.*", + jetstream_server + ); + let subscriber = ScrobbleSubscriber::new(&url); + + subscriber.run().await?; + + Ok(()) +} diff --git a/crates/jetstream/src/main.rs b/crates/jetstream/src/main.rs index 595d5c1b..efdb0daf 100644 --- a/crates/jetstream/src/main.rs +++ b/crates/jetstream/src/main.rs @@ -13,7 +13,7 @@ pub mod xata; async fn main() -> Result<(), anyhow::Error> { dotenv()?; let jetstream_server = env::var("JETSTREAM_SERVER") - .unwrap_or_else(|_| "wss://jetstream2.us-east.bsky.network".to_string()); + .unwrap_or_else(|_| "wss://jetstream2.us-west.bsky.network".to_string()); let url = format!( "{}/subscribe?wantedCollections=app.rocksky.*", jetstream_server diff --git a/crates/playlists/Cargo.toml b/crates/playlists/Cargo.toml index af2de4c6..b2f5aac9 100644 --- a/crates/playlists/Cargo.toml +++ b/crates/playlists/Cargo.toml @@ -1,5 +1,5 @@ [package] -name = "playlists" +name = "rocksky-playlists" version = "0.1.0" authors.workspace = true edition.workspace = true diff --git a/crates/playlists/src/core.rs b/crates/playlists/src/core.rs index 6e47b996..db6ce8ed 100644 --- a/crates/playlists/src/core.rs +++ b/crates/playlists/src/core.rs @@ -14,7 +14,7 @@ use sqlx::{Pool, Postgres}; use crate::{ crypto::{decrypt_aes_256_ctr, generate_token}, types::{self, spotify_token::SpotifyTokenWithEmail}, - xata::{self, track::Track}, + xata::{self}, }; const ROCKSKY_API: &str = "https://api.rocksky.app"; diff --git a/crates/playlists/src/lib.rs b/crates/playlists/src/lib.rs index ff984676..4bdc65e2 100644 --- a/crates/playlists/src/lib.rs +++ b/crates/playlists/src/lib.rs @@ -1,6 +1,65 @@ +use std::{ + env, + sync::{Arc, Mutex}, +}; + +use anyhow::Error; +use async_nats::connect; +use duckdb::Connection; +use owo_colors::OwoColorize; +use sqlx::postgres::PgPoolOptions; + +use crate::{ + core::{create_tables, find_spotify_users, load_users, save_playlists}, + spotify::get_user_playlists, + subscriber::subscribe, +}; + pub mod core; pub mod crypto; pub mod spotify; pub mod subscriber; pub mod types; pub mod xata; + +pub async fn start() -> Result<(), Error> { + let conn = Connection::open("./rocksky-playlists.ddb")?; + let conn = Arc::new(Mutex::new(conn)); + create_tables(conn.clone())?; + + subscribe(conn.clone()).await?; + + let pool = PgPoolOptions::new() + .max_connections(5) + .connect(&env::var("XATA_POSTGRES_URL")?) + .await?; + let users = find_spotify_users(&pool, 0, 100).await?; + + load_users(conn.clone(), &pool).await?; + + sqlx::query(r#" + CREATE UNIQUE INDEX IF NOT EXISTS user_playlists_unique_index ON user_playlists (user_id, playlist_id) + "#) + .execute(&pool) + .await?; + let conn = conn.clone(); + + let addr = env::var("NATS_URL").unwrap_or_else(|_| "nats://localhost:4222".to_string()); + let nc = connect(&addr).await?; + let nc = Arc::new(Mutex::new(nc)); + println!("Connected to NATS server at {}", addr.bright_green()); + + for user in users { + let token = user.1.clone(); + let did = user.2.clone(); + let user_id = user.3.clone(); + let playlists = get_user_playlists(token).await?; + save_playlists(&pool, conn.clone(), nc.clone(), playlists, &user_id, &did).await?; + } + + println!("Done!"); + + loop { + tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; + } +} diff --git a/crates/playlists/src/main.rs b/crates/playlists/src/main.rs index 02b32f98..5b845cd7 100644 --- a/crates/playlists/src/main.rs +++ b/crates/playlists/src/main.rs @@ -9,7 +9,7 @@ use async_nats::connect; use dotenv::dotenv; use duckdb::Connection; use owo_colors::OwoColorize; -use playlists::subscriber::subscribe; +use rocksky_playlists::subscriber::subscribe; use spotify::get_user_playlists; use sqlx::postgres::PgPoolOptions; diff --git a/crates/rockskyd/Cargo.toml b/crates/rockskyd/Cargo.toml new file mode 100644 index 00000000..66eb6389 --- /dev/null +++ b/crates/rockskyd/Cargo.toml @@ -0,0 +1,22 @@ +[package] +name = "rockskyd" +version = "0.1.0" +authors.workspace = true +edition.workspace = true +license.workspace = true +repository.workspace = true + +[dependencies] +clap = "4.5.31" +tokio = { version = "1.43.0", features = ["full"] } +dotenv = "0.15.0" +anyhow = "1.0.96" +rocksky-analytics = { path = "../analytics" } +rocksky-dropbox = { path = "../dropbox" } +rocksky-googledrive = { path = "../googledrive" } +rocksky-jetstream = { path = "../jetstream" } +rocksky-playlists = { path = "../playlists" } +rocksky-scrobbler = { path = "../scrobbler" } +rocksky-spotify = { path = "../spotify" } +rocksky-tracklist = { path = "../tracklist" } +rocksky-webscrobbler = { path = "../webscrobbler" } diff --git a/crates/rockskyd/src/cmd/analytics.rs b/crates/rockskyd/src/cmd/analytics.rs new file mode 100644 index 00000000..68a64a75 --- /dev/null +++ b/crates/rockskyd/src/cmd/analytics.rs @@ -0,0 +1,11 @@ +use anyhow::Error; + +pub async fn sync() -> Result<(), Error> { + rocksky_analytics::sync().await?; + Ok(()) +} + +pub async fn serve() -> Result<(), Error> { + rocksky_analytics::serve().await?; + Ok(()) +} diff --git a/crates/rockskyd/src/cmd/dropbox.rs b/crates/rockskyd/src/cmd/dropbox.rs new file mode 100644 index 00000000..106349b1 --- /dev/null +++ b/crates/rockskyd/src/cmd/dropbox.rs @@ -0,0 +1,11 @@ +use anyhow::Error; + +pub async fn scan() -> Result<(), Error> { + rocksky_dropbox::cmd::scan::scan().await?; + Ok(()) +} + +pub async fn serve() -> Result<(), Error> { + rocksky_dropbox::cmd::serve::serve().await?; + Ok(()) +} diff --git a/crates/rockskyd/src/cmd/googledrive.rs b/crates/rockskyd/src/cmd/googledrive.rs new file mode 100644 index 00000000..8b16ca72 --- /dev/null +++ b/crates/rockskyd/src/cmd/googledrive.rs @@ -0,0 +1,11 @@ +use anyhow::Error; + +pub async fn scan() -> Result<(), Error> { + rocksky_googledrive::cmd::scan::scan().await?; + Ok(()) +} + +pub async fn serve() -> Result<(), Error> { + rocksky_googledrive::cmd::serve::serve().await?; + Ok(()) +} diff --git a/crates/rockskyd/src/cmd/jetstream.rs b/crates/rockskyd/src/cmd/jetstream.rs new file mode 100644 index 00000000..a5f4be28 --- /dev/null +++ b/crates/rockskyd/src/cmd/jetstream.rs @@ -0,0 +1,6 @@ +use anyhow::Error; + +pub async fn start_jetstream_service() -> Result<(), Error> { + rocksky_jetstream::subscribe().await?; + Ok(()) +} diff --git a/crates/rockskyd/src/cmd/mod.rs b/crates/rockskyd/src/cmd/mod.rs new file mode 100644 index 00000000..74b52f72 --- /dev/null +++ b/crates/rockskyd/src/cmd/mod.rs @@ -0,0 +1,9 @@ +pub mod analytics; +pub mod dropbox; +pub mod googledrive; +pub mod jetstream; +pub mod playlist; +pub mod scrobbler; +pub mod spotify; +pub mod tracklist; +pub mod webscrobbler; diff --git a/crates/rockskyd/src/cmd/playlist.rs b/crates/rockskyd/src/cmd/playlist.rs new file mode 100644 index 00000000..deb26856 --- /dev/null +++ b/crates/rockskyd/src/cmd/playlist.rs @@ -0,0 +1,6 @@ +use anyhow::Error; + +pub async fn start_playlist_service() -> Result<(), Error> { + rocksky_playlists::start().await?; + Ok(()) +} diff --git a/crates/rockskyd/src/cmd/scrobbler.rs b/crates/rockskyd/src/cmd/scrobbler.rs new file mode 100644 index 00000000..cf15d654 --- /dev/null +++ b/crates/rockskyd/src/cmd/scrobbler.rs @@ -0,0 +1,6 @@ +use anyhow::Error; + +pub async fn start_scrobbler_service() -> Result<(), Error> { + rocksky_scrobbler::run().await?; + Ok(()) +} diff --git a/crates/rockskyd/src/cmd/spotify.rs b/crates/rockskyd/src/cmd/spotify.rs new file mode 100644 index 00000000..ddbceb2e --- /dev/null +++ b/crates/rockskyd/src/cmd/spotify.rs @@ -0,0 +1,6 @@ +use anyhow::Error; + +pub async fn start_spotify_service() -> Result<(), Error> { + rocksky_spotify::run().await?; + Ok(()) +} diff --git a/crates/rockskyd/src/cmd/tracklist.rs b/crates/rockskyd/src/cmd/tracklist.rs new file mode 100644 index 00000000..e33f34c3 --- /dev/null +++ b/crates/rockskyd/src/cmd/tracklist.rs @@ -0,0 +1,6 @@ +use anyhow::Error; + +pub async fn start_tracklist_service() -> Result<(), Error> { + rocksky_tracklist::run().await?; + Ok(()) +} diff --git a/crates/rockskyd/src/cmd/webscrobbler.rs b/crates/rockskyd/src/cmd/webscrobbler.rs new file mode 100644 index 00000000..49dededa --- /dev/null +++ b/crates/rockskyd/src/cmd/webscrobbler.rs @@ -0,0 +1,6 @@ +use anyhow::Error; + +pub async fn start_webscrobbler_service() -> Result<(), Error> { + rocksky_webscrobbler::start_server().await?; + Ok(()) +} diff --git a/crates/rockskyd/src/main.rs b/crates/rockskyd/src/main.rs new file mode 100644 index 00000000..ef07ac1f --- /dev/null +++ b/crates/rockskyd/src/main.rs @@ -0,0 +1,82 @@ +use clap::Command; +use dotenv::dotenv; + +pub mod cmd; + +fn cli() -> Command { + Command::new("rockskyd") + .version(env!("CARGO_PKG_VERSION")) + .about("Rocksky Daemon Service") + .subcommand( + Command::new("analytics") + .about("Analytics related commands") + .subcommand(Command::new("sync").about("Sync data from Xata to DuckDB")) + .subcommand(Command::new("serve").about("Serve the Rocksky Analytics API")), + ) + .subcommand( + Command::new("dropbox") + .about("Dropbox related commands") + .subcommand(Command::new("scan").about("Scan Dropbox Music Folder")) + .subcommand(Command::new("serve").about("Serve Rocksky Dropbox API")), + ) + .subcommand( + Command::new("googledrive") + .about("Google Drive related commands") + .subcommand(Command::new("scan").about("Scan Google Drive Music Folder")) + .subcommand(Command::new("serve").about("Serve Rocksky Google Drive API")), + ) + .subcommand(Command::new("jetstream").about("Start JetStream Subscriber Service")) + .subcommand(Command::new("playlist").about("Playlist related commands")) + .subcommand(Command::new("scrobbler").about("Start Scrobbler API")) + .subcommand(Command::new("spotify").about("Start Spotify Listener Service")) + .subcommand(Command::new("tracklist").about("Start User Current Track Queue Service")) + .subcommand(Command::new("webscrobbler").about("Start Webscrobbler API")) +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + dotenv().ok(); + + let args = cli().get_matches(); + + match args.subcommand() { + Some(("analytics", sub_m)) => match sub_m.subcommand() { + Some(("sync", _)) => cmd::analytics::sync().await?, + Some(("serve", _)) => cmd::analytics::serve().await?, + _ => println!("Unknown analytics command"), + }, + Some(("dropbox", sub_m)) => match sub_m.subcommand() { + Some(("scan", _)) => cmd::dropbox::scan().await?, + Some(("serve", _)) => cmd::dropbox::serve().await?, + _ => println!("Unknown dropbox command"), + }, + Some(("googledrive", sub_m)) => match sub_m.subcommand() { + Some(("scan", _)) => cmd::googledrive::scan().await?, + Some(("serve", _)) => cmd::googledrive::serve().await?, + _ => println!("Unknown googledrive command"), + }, + Some(("jetstream", _)) => { + cmd::jetstream::start_jetstream_service().await?; + } + Some(("playlist", _)) => { + cmd::playlist::start_playlist_service().await?; + } + Some(("scrobbler", _)) => { + cmd::scrobbler::start_scrobbler_service().await?; + } + Some(("spotify", _)) => { + cmd::spotify::start_spotify_service().await?; + } + Some(("tracklist", _)) => { + cmd::tracklist::start_tracklist_service().await?; + } + Some(("webscrobbler", _)) => { + cmd::webscrobbler::start_webscrobbler_service().await?; + } + _ => { + println!("No valid subcommand was used. Use --help to see available commands."); + } + } + + Ok(()) +} diff --git a/crates/scrobbler/Cargo.toml b/crates/scrobbler/Cargo.toml index fa8035df..dc5da3ae 100644 --- a/crates/scrobbler/Cargo.toml +++ b/crates/scrobbler/Cargo.toml @@ -1,5 +1,5 @@ [package] -name = "scrobbler" +name = "rocksky-scrobbler" version = "0.1.0" authors.workspace = true edition.workspace = true diff --git a/crates/scrobbler/src/lib.rs b/crates/scrobbler/src/lib.rs new file mode 100644 index 00000000..e615d465 --- /dev/null +++ b/crates/scrobbler/src/lib.rs @@ -0,0 +1,103 @@ +pub mod auth; +pub mod cache; +pub mod crypto; +pub mod handlers; +pub mod listenbrainz; +pub mod musicbrainz; +pub mod params; +pub mod repo; +pub mod response; +pub mod rocksky; +pub mod scrobbler; +pub mod signature; +pub mod spotify; +pub mod types; +pub mod xata; + +use std::{env, sync::Arc, time::Duration}; + +use actix_limitation::{Limiter, RateLimiter}; +use actix_session::SessionExt; +use actix_web::{ + dev::ServiceRequest, + web::{self, Data}, + App, HttpServer, +}; +use anyhow::Error; +use owo_colors::OwoColorize; +use sqlx::postgres::PgPoolOptions; + +use crate::cache::Cache; + +pub const BANNER: &str = r#" + ___ ___ _____ __ __ __ + / | __ ______/ (_)___ / ___/______________ / /_ / /_ / /__ _____ + / /| |/ / / / __ / / __ \ \__ \/ ___/ ___/ __ \/ __ \/ __ \/ / _ \/ ___/ + / ___ / /_/ / /_/ / / /_/ / ___/ / /__/ / / /_/ / /_/ / /_/ / / __/ / +/_/ |_\__,_/\__,_/_/\____/ /____/\___/_/ \____/_.___/_.___/_/\___/_/ + + This is the Rocksky Scrobbler API compatible with Last.fm AudioScrobbler API +"#; + +pub async fn run() -> Result<(), Error> { + println!("{}", BANNER.magenta()); + + let cache = Cache::new()?; + + let pool = PgPoolOptions::new() + .max_connections(5) + .connect(&env::var("XATA_POSTGRES_URL")?) + .await?; + let conn = Arc::new(pool); + + let host = env::var("SCROBBLE_HOST").unwrap_or_else(|_| "127.0.0.1".to_string()); + let port = env::var("SCROBBLE_PORT") + .unwrap_or_else(|_| "7882".to_string()) + .parse::() + .unwrap_or(7882); + + println!( + "Starting Scrobble server @ {}", + format!("{}:{}", host, port).green() + ); + + let limiter = web::Data::new( + Limiter::builder("redis://127.0.0.1") + .key_by(|req: &ServiceRequest| { + req.get_session() + .get(&"session-id") + .unwrap_or_else(|_| req.cookie(&"rate-api-id").map(|c| c.to_string())) + }) + .limit(100) + .period(Duration::from_secs(60)) // 60 minutes + .build() + .unwrap(), + ); + + HttpServer::new(move || { + App::new() + .wrap(RateLimiter::default()) + .app_data(limiter.clone()) + .app_data(Data::new(conn.clone())) + .app_data(Data::new(cache.clone())) + .service(handlers::handle_methods) + .service(handlers::handle_nowplaying) + .service(handlers::handle_submission) + .service(listenbrainz::handlers::handle_submit_listens) + .service(listenbrainz::handlers::handle_validate_token) + .service(listenbrainz::handlers::handle_search_users) + .service(listenbrainz::handlers::handle_get_playing_now) + .service(listenbrainz::handlers::handle_get_listens) + .service(listenbrainz::handlers::handle_get_listen_count) + .service(listenbrainz::handlers::handle_get_artists) + .service(listenbrainz::handlers::handle_get_recordings) + .service(listenbrainz::handlers::handle_get_release_groups) + .service(handlers::index) + .service(handlers::handle_get) + }) + .bind((host, port))? + .run() + .await?; + + Ok(()) +} diff --git a/crates/spotify/Cargo.toml b/crates/spotify/Cargo.toml index 94b7c896..6f778df8 100644 --- a/crates/spotify/Cargo.toml +++ b/crates/spotify/Cargo.toml @@ -1,5 +1,5 @@ [package] -name = "spotify" +name = "rocksky-spotify" version = "0.1.0" authors.workspace = true edition.workspace = true diff --git a/crates/spotify/src/lib.rs b/crates/spotify/src/lib.rs new file mode 100644 index 00000000..590cdc8c --- /dev/null +++ b/crates/spotify/src/lib.rs @@ -0,0 +1,897 @@ +use std::{ + collections::HashMap, + env, + sync::{atomic::AtomicBool, Arc, Mutex}, + thread, +}; + +use anyhow::Error; +use async_nats::connect; +use owo_colors::OwoColorize; +use reqwest::Client; +use sqlx::{postgres::PgPoolOptions, Pool, Postgres}; +use tokio_stream::StreamExt; + +use crate::{ + cache::Cache, + crypto::decrypt_aes_256_ctr, + rocksky::{scrobble, update_library}, + types::{ + album_tracks::AlbumTracks, + currently_playing::{Album, Artist, CurrentlyPlaying}, + spotify_token::SpotifyTokenWithEmail, + token::AccessToken, + }, +}; + +pub mod cache; +pub mod crypto; +pub mod rocksky; +pub mod token; +pub mod types; + +pub const BASE_URL: &str = "https://spotify-api.rocksky.app/v1"; + +pub async fn run() -> Result<(), Error> { + let cache = Cache::new()?; + let pool = PgPoolOptions::new() + .max_connections(5) + .connect(&env::var("XATA_POSTGRES_URL")?) + .await?; + + let addr = env::var("NATS_URL").unwrap_or_else(|_| "nats://localhost:4222".to_string()); + let nc = connect(&addr).await?; + println!("Connected to NATS server at {}", addr.bright_green()); + + let mut sub = nc.subscribe("rocksky.spotify.user".to_string()).await?; + println!("Subscribed to {}", "rocksky.spotify.user".bright_green()); + + let users = find_spotify_users(&pool, 0, 100).await?; + println!("Found {} users", users.len().bright_green()); + + // Shared HashMap to manage threads and their stop flags + let thread_map: Arc>>> = + Arc::new(Mutex::new(HashMap::new())); + + // Start threads for all users + for user in users { + let email = user.0.clone(); + let token = user.1.clone(); + let did = user.2.clone(); + let stop_flag = Arc::new(AtomicBool::new(false)); + let cache = cache.clone(); + let nc = nc.clone(); + let thread_map = Arc::clone(&thread_map); + + thread_map + .lock() + .unwrap() + .insert(email.clone(), Arc::clone(&stop_flag)); + + thread::spawn(move || { + let rt = tokio::runtime::Runtime::new().unwrap(); + match rt.block_on(async { + watch_currently_playing(email.clone(), token, did, stop_flag, cache.clone()) + .await?; + Ok::<(), Error>(()) + }) { + Ok(_) => {} + Err(e) => { + println!( + "{} Error starting thread for user: {} - {}", + format!("[{}]", email).bright_green(), + email.bright_green(), + e.to_string().bright_red() + ); + + // If there's an error, publish a message to restart the thread + match rt.block_on(nc.publish("rocksky.spotify.user", email.clone().into())) { + Ok(_) => { + println!( + "{} Published message to restart thread for user: {}", + format!("[{}]", email).bright_green(), + email.bright_green() + ); + } + Err(e) => { + println!( + "{} Error publishing message to restart thread: {}", + format!("[{}]", email).bright_green(), + e.to_string().bright_red() + ); + } + } + } + } + }); + } + + // Handle subscription messages + while let Some(message) = sub.next().await { + let user_id = String::from_utf8(message.payload.to_vec()).unwrap(); + println!( + "Received message to restart thread for user: {}", + user_id.bright_green() + ); + + let mut thread_map = thread_map.lock().unwrap(); + + // Check if the user exists in the thread map + if let Some(stop_flag) = thread_map.get(&user_id) { + // Stop the existing thread + stop_flag.store(true, std::sync::atomic::Ordering::Relaxed); + + // Create a new stop flag and restart the thread + let new_stop_flag = Arc::new(AtomicBool::new(false)); + thread_map.insert(user_id.clone(), Arc::clone(&new_stop_flag)); + + let user = find_spotify_user(&pool, &user_id).await?; + + if user.is_none() { + println!( + "Spotify user not found: {}, skipping", + user_id.bright_green() + ); + continue; + } + + let user = user.unwrap(); + + let email = user.0.clone(); + let token = user.1.clone(); + let did = user.2.clone(); + let cache = cache.clone(); + + thread::spawn(move || { + let rt = tokio::runtime::Runtime::new().unwrap(); + match rt.block_on(async { + watch_currently_playing( + email.clone(), + token, + did, + new_stop_flag, + cache.clone(), + ) + .await?; + Ok::<(), Error>(()) + }) { + Ok(_) => {} + Err(e) => { + println!( + "{} Error restarting thread for user: {} - {}", + format!("[{}]", email).bright_green(), + email.bright_green(), + e.to_string().bright_red() + ); + } + } + }); + + println!("Restarted thread for user: {}", user_id.bright_green()); + } else { + println!( + "No thread found for user: {}, starting new thread", + user_id.bright_green() + ); + let user = find_spotify_user(&pool, &user_id).await?; + if let Some(user) = user { + let email = user.0.clone(); + let token = user.1.clone(); + let did = user.2.clone(); + let stop_flag = Arc::new(AtomicBool::new(false)); + let cache = cache.clone(); + let nc = nc.clone(); + + thread_map.insert(email.clone(), Arc::clone(&stop_flag)); + + thread::spawn(move || { + let rt = tokio::runtime::Runtime::new().unwrap(); + match rt.block_on(async { + watch_currently_playing( + email.clone(), + token, + did, + stop_flag, + cache.clone(), + ) + .await?; + Ok::<(), Error>(()) + }) { + Ok(_) => {} + Err(e) => { + println!( + "{} Error starting thread for user: {} - {}", + format!("[{}]", email).bright_green(), + email.bright_green(), + e.to_string().bright_red() + ); + match rt + .block_on(nc.publish("rocksky.spotify.user", email.clone().into())) + { + Ok(_) => {} + Err(e) => { + println!( + "{} Error publishing message to restart thread: {}", + format!("[{}]", email).bright_green(), + e.to_string().bright_red() + ); + } + } + } + } + }); + } + } + } + + Ok(()) +} + +pub async fn refresh_token(token: &str) -> Result { + if env::var("SPOTIFY_CLIENT_ID").is_err() || env::var("SPOTIFY_CLIENT_SECRET").is_err() { + panic!("Please set SPOTIFY_CLIENT_ID and SPOTIFY_CLIENT_SECRET environment variables"); + } + + let client_id = env::var("SPOTIFY_CLIENT_ID")?; + let client_secret = env::var("SPOTIFY_CLIENT_SECRET")?; + + let client = Client::new(); + + let response = client + .post("https://accounts.spotify.com/api/token") + .basic_auth(&client_id, Some(client_secret)) + .form(&[ + ("grant_type", "refresh_token"), + ("refresh_token", token), + ("client_id", &client_id), + ]) + .send() + .await?; + let token = response.json::().await?; + Ok(token) +} + +pub async fn get_currently_playing( + cache: Cache, + user_id: &str, + token: &str, +) -> Result, Error> { + if let Ok(Some(data)) = cache.get(user_id) { + println!( + "{} {}", + format!("[{}]", user_id).bright_green(), + "Using cache".cyan() + ); + if data == "No content" { + return Ok(None); + } + let decoded_data = serde_json::from_str::(&data); + + if decoded_data.is_err() { + println!( + "{} {} {}", + format!("[{}]", user_id).bright_green(), + "Cache is invalid".red(), + data + ); + cache.setex(user_id, "No content", 10)?; + cache.del(&format!("{}:current", user_id))?; + return Ok(None); + } + + let data: CurrentlyPlaying = decoded_data.unwrap(); + // detect if the song has changed + let previous = cache.get(&format!("{}:previous", user_id)); + + if previous.is_err() { + println!( + "{} redis error: {}", + format!("[{}]", user_id).bright_green(), + previous.unwrap_err().to_string().bright_red() + ); + return Ok(None); + } + + let previous = previous.unwrap(); + + let changed = match previous { + Some(previous) => { + if serde_json::from_str::(&previous).is_err() { + println!( + "{} {} {}", + format!("[{}]", user_id).bright_green(), + "Previous cache is invalid", + previous + ); + return Ok(None); + } + + let previous: CurrentlyPlaying = serde_json::from_str(&previous)?; + if previous.item.is_none() && data.item.is_some() { + return Ok(Some((data, true))); + } + + if previous.item.is_some() && data.item.is_none() { + return Ok(Some((data, false))); + } + + if previous.item.is_none() && data.item.is_none() { + return Ok(Some((data, false))); + } + + let previous_item = previous.item.unwrap(); + let data_item = data.clone().item.unwrap(); + previous_item.id != data_item.id + && previous.progress_ms.unwrap_or(0) != data.progress_ms.unwrap_or(0) + } + _ => true, + }; + return Ok(Some((data, changed))); + } + + let token = refresh_token(token).await?; + let client = Client::new(); + let response = client + .get(format!("{}/me/player/currently-playing", BASE_URL)) + .bearer_auth(token.access_token) + .send() + .await?; + + let headers = response.headers().clone(); + let status = response.status().as_u16(); + let data = response.text().await?; + + if status == 429 { + println!( + "{} Too many requests, retry-after {}", + format!("[{}]", user_id).bright_green(), + headers + .get("retry-after") + .unwrap() + .to_str() + .unwrap() + .bright_green() + ); + return Ok(None); + } + + let previous = cache.get(&format!("{}:previous", user_id)); + if previous.is_err() { + println!( + "{} redis error: {}", + format!("[{}]", user_id).bright_green(), + previous.unwrap_err().to_string().bright_red() + ); + return Ok(None); + } + + let previous = previous.unwrap(); + + // check if status code is 204 + if status == 204 { + println!("No content"); + match cache.setex( + user_id, + "No content", + match previous.is_none() { + true => 30, + false => 10, + }, + ) { + Ok(_) => {} + Err(e) => { + println!( + "{} redis error: {}", + format!("[{}]", user_id).bright_green(), + e.to_string().bright_red() + ); + return Ok(None); + } + } + match cache.del(&format!("{}:current", user_id)) { + Ok(_) => {} + Err(e) => { + println!( + "{} redis error: {}", + format!("[{}]", user_id).bright_green(), + e.to_string().bright_red() + ); + return Ok(None); + } + } + return Ok(None); + } + + if serde_json::from_str::(&data).is_err() { + println!( + "{} {} {}", + format!("[{}]", user_id).bright_green(), + "Invalid data received".red(), + data + ); + match cache.setex(user_id, "No content", 10) { + Ok(_) => {} + Err(e) => { + println!( + "{} redis error: {}", + format!("[{}]", user_id).bright_green(), + e.to_string().bright_red() + ); + return Ok(None); + } + } + match cache.del(&format!("{}:current", user_id)) { + Ok(_) => {} + Err(e) => { + println!( + "{} redis error: {}", + format!("[{}]", user_id).bright_green(), + e.to_string().bright_red() + ); + return Ok(None); + } + } + return Ok(None); + } + + let data = serde_json::from_str::(&data)?; + + match cache.setex( + user_id, + &serde_json::to_string(&data)?, + match previous.is_none() { + true => 30, + false => 15, + }, + ) { + Ok(_) => {} + Err(e) => { + println!( + "{} redis error: {}", + format!("[{}]", user_id).bright_green(), + e.to_string().bright_red() + ); + return Ok(None); + } + } + match cache.del(&format!("{}:current", user_id)) { + Ok(_) => {} + Err(e) => { + println!( + "{} redis error: {}", + format!("[{}]", user_id).bright_green(), + e.to_string().bright_red() + ); + return Ok(None); + } + } + + // detect if the song has changed + let previous = cache.get(&format!("{}:previous", user_id)); + + if previous.is_err() { + println!( + "{} redis error: {}", + format!("[{}]", user_id).bright_green(), + previous.unwrap_err().to_string().bright_red() + ); + return Ok(None); + } + + let previous = previous.unwrap(); + let changed = match previous { + Some(previous) => { + if serde_json::from_str::(&previous).is_err() { + println!( + "{} {} {}", + format!("[{}]", user_id).bright_green(), + "Previous cache is invalid", + previous + ); + return Ok(None); + } + + let previous: CurrentlyPlaying = serde_json::from_str(&previous)?; + if previous.item.is_none() || data.item.is_none() { + return Ok(Some((data, false))); + } + + let previous_item = previous.item.unwrap(); + let data_item = data.clone().item.unwrap(); + + previous_item.id != data_item.id + && previous.progress_ms.unwrap_or(0) != data.progress_ms.unwrap_or(0) + } + _ => false, + }; + + // save as previous song + match cache.setex( + &format!("{}:previous", user_id), + &serde_json::to_string(&data)?, + 600, + ) { + Ok(_) => {} + Err(e) => { + println!( + "{} redis error: {}", + format!("[{}]", user_id).bright_green(), + e.to_string().bright_red() + ); + return Ok(None); + } + } + + Ok(Some((data, changed))) +} + +pub async fn get_artist( + cache: Cache, + artist_id: &str, + token: &str, +) -> Result, Error> { + if let Ok(Some(data)) = cache.get(artist_id) { + return Ok(Some(serde_json::from_str(&data)?)); + } + + let token = refresh_token(token).await?; + let client = Client::new(); + let response = client + .get(&format!("{}/artists/{}", BASE_URL, artist_id)) + .bearer_auth(token.access_token) + .send() + .await?; + + let headers = response.headers().clone(); + let data = response.text().await?; + + if data == "Too many requests" { + println!( + "> retry-after {}", + headers.get("retry-after").unwrap().to_str().unwrap() + ); + println!("> {} [get_artist]", data); + return Ok(None); + } + + match cache.setex(artist_id, &data, 20) { + Ok(_) => {} + Err(e) => { + println!( + "{} redis error: {}", + format!("[{}]", artist_id).bright_green(), + e.to_string().bright_red() + ); + return Ok(None); + } + } + + Ok(Some(serde_json::from_str(&data)?)) +} + +pub async fn get_album(cache: Cache, album_id: &str, token: &str) -> Result, Error> { + if let Ok(Some(data)) = cache.get(album_id) { + return Ok(Some(serde_json::from_str(&data)?)); + } + + let token = refresh_token(token).await?; + let client = Client::new(); + let response = client + .get(&format!("{}/albums/{}", BASE_URL, album_id)) + .bearer_auth(token.access_token) + .send() + .await?; + + let headers = response.headers().clone(); + let data = response.text().await?; + + if data == "Too many requests" { + println!( + "> retry-after {}", + headers.get("retry-after").unwrap().to_str().unwrap() + ); + println!("> {} [get_album]", data); + return Ok(None); + } + + match cache.setex(album_id, &data, 20) { + Ok(_) => {} + Err(e) => { + println!( + "{} redis error: {}", + format!("[{}]", album_id).bright_green(), + e.to_string().bright_red() + ); + return Ok(None); + } + } + + Ok(Some(serde_json::from_str(&data)?)) +} + +pub async fn get_album_tracks( + cache: Cache, + album_id: &str, + token: &str, +) -> Result { + if let Ok(Some(data)) = cache.get(&format!("{}:tracks", album_id)) { + return Ok(serde_json::from_str(&data)?); + } + + let token = refresh_token(token).await?; + let client = Client::new(); + let mut all_tracks = Vec::new(); + let mut offset = 0; + let limit = 50; + + loop { + let response = client + .get(&format!("{}/albums/{}/tracks", BASE_URL, album_id)) + .bearer_auth(&token.access_token) + .query(&[ + ("limit", &limit.to_string()), + ("offset", &offset.to_string()), + ]) + .send() + .await?; + + let headers = response.headers().clone(); + let data = response.text().await?; + if data == "Too many requests" { + println!( + "> retry-after {}", + headers.get("retry-after").unwrap().to_str().unwrap() + ); + println!("> {} [get_album_tracks]", data); + continue; + } + + let album_tracks: AlbumTracks = serde_json::from_str(&data)?; + + if album_tracks.items.is_empty() { + break; + } + + all_tracks.extend(album_tracks.items); + offset += limit; + } + + let all_tracks_json = serde_json::to_string(&all_tracks)?; + match cache.setex(&format!("{}:tracks", album_id), &all_tracks_json, 20) { + Ok(_) => {} + Err(e) => { + println!( + "{} redis error: {}", + format!("[{}]", album_id).bright_green(), + e.to_string().bright_red() + ); + } + } + + Ok(AlbumTracks { + items: all_tracks, + ..Default::default() + }) +} + +pub async fn find_spotify_users( + pool: &Pool, + offset: usize, + limit: usize, +) -> Result, Error> { + let results: Vec = sqlx::query_as( + r#" + SELECT * FROM spotify_tokens + LEFT JOIN spotify_accounts ON spotify_tokens.user_id = spotify_accounts.user_id + LEFT JOIN users ON spotify_accounts.user_id = users.xata_id + LIMIT $1 OFFSET $2 + "#, + ) + .bind(limit as i64) + .bind(offset as i64) + .fetch_all(pool) + .await?; + + let mut user_tokens = vec![]; + + for result in &results { + let token = decrypt_aes_256_ctr( + &result.refresh_token, + &hex::decode(env::var("SPOTIFY_ENCRYPTION_KEY")?)?, + )?; + user_tokens.push(( + result.email.clone(), + token, + result.did.clone(), + result.user_id.clone(), + )); + } + + Ok(user_tokens) +} + +pub async fn find_spotify_user( + pool: &Pool, + email: &str, +) -> Result, Error> { + let result: Vec = sqlx::query_as( + r#" + SELECT * FROM spotify_tokens + LEFT JOIN spotify_accounts ON spotify_tokens.user_id = spotify_accounts.user_id + LEFT JOIN users ON spotify_accounts.user_id = users.xata_id + WHERE spotify_accounts.email = $1 + "#, + ) + .bind(email) + .fetch_all(pool) + .await?; + + match result.first() { + Some(result) => { + let token = decrypt_aes_256_ctr( + &result.refresh_token, + &hex::decode(env::var("SPOTIFY_ENCRYPTION_KEY")?)?, + )?; + Ok(Some((result.email.clone(), token, result.did.clone()))) + } + None => Ok(None), + } +} + +pub async fn watch_currently_playing( + spotify_email: String, + token: String, + did: String, + stop_flag: Arc, + cache: Cache, +) -> Result<(), Error> { + println!( + "{} {}", + format!("[{}]", spotify_email).bright_green(), + "Checking currently playing".cyan() + ); + + let stop_flag_clone = stop_flag.clone(); + let spotify_email_clone = spotify_email.clone(); + let cache_clone = cache.clone(); + thread::spawn(move || { + loop { + if stop_flag_clone.load(std::sync::atomic::Ordering::Relaxed) { + println!( + "{} Stopping Thread", + format!("[{}]", spotify_email_clone).bright_green() + ); + break; + } + if let Ok(Some(cached)) = cache_clone.get(&format!("{}:current", spotify_email_clone)) { + if serde_json::from_str::(&cached).is_err() { + thread::sleep(std::time::Duration::from_millis(800)); + continue; + } + + let mut current_song = serde_json::from_str::(&cached)?; + + if let Some(item) = current_song.item.clone() { + if current_song.is_playing + && current_song.progress_ms.unwrap_or(0) < item.duration_ms.into() + { + current_song.progress_ms = + Some(current_song.progress_ms.unwrap_or(0) + 800); + match cache_clone.setex( + &format!("{}:current", spotify_email_clone), + &serde_json::to_string(¤t_song)?, + 16, + ) { + Ok(_) => {} + Err(e) => { + println!( + "{} redis error: {}", + format!("[{}]", spotify_email_clone).bright_green(), + e.to_string().bright_red() + ); + } + } + thread::sleep(std::time::Duration::from_millis(800)); + continue; + } + } + continue; + } + + if let Ok(Some(cached)) = cache_clone.get(&spotify_email_clone) { + if cached == "No content" { + thread::sleep(std::time::Duration::from_millis(800)); + continue; + } + match cache_clone.setex(&format!("{}:current", spotify_email_clone), &cached, 16) { + Ok(_) => {} + Err(e) => { + println!( + "{} redis error: {}", + format!("[{}]", spotify_email_clone).bright_green(), + e.to_string().bright_red() + ); + } + } + } + + thread::sleep(std::time::Duration::from_millis(800)); + } + Ok::<(), Error>(()) + }); + + loop { + if stop_flag.load(std::sync::atomic::Ordering::Relaxed) { + println!( + "{} Stopping Thread", + format!("[{}]", spotify_email).bright_green() + ); + break; + } + let spotify_email = spotify_email.clone(); + let token = token.clone(); + let did = did.clone(); + let cache = cache.clone(); + + let currently_playing = get_currently_playing(cache.clone(), &spotify_email, &token).await; + let currently_playing = match currently_playing { + Ok(currently_playing) => currently_playing, + Err(e) => { + println!( + "{} {}", + format!("[{}]", spotify_email).bright_green(), + e.to_string().bright_red() + ); + tokio::time::sleep(tokio::time::Duration::from_millis(1000)).await; + continue; + } + }; + + if let Some((data, changed)) = currently_playing { + if data.item.is_none() { + println!( + "{} {}", + format!("[{}]", spotify_email).bright_green(), + "No song playing".yellow() + ); + tokio::time::sleep(tokio::time::Duration::from_millis(1000)).await; + continue; + } + let data_item = data.item.unwrap(); + println!( + "{} {} is_playing: {} changed: {}", + format!("[{}]", spotify_email).bright_green(), + format!("{} - {}", data_item.name, data_item.artists[0].name).yellow(), + data.is_playing, + changed + ); + + if changed { + scrobble(cache.clone(), &spotify_email, &did, &token).await?; + + thread::spawn(move || { + let rt = tokio::runtime::Runtime::new().unwrap(); + match rt.block_on(async { + get_album_tracks(cache.clone(), &data_item.album.id, &token).await?; + get_album(cache.clone(), &data_item.album.id, &token).await?; + update_library(cache.clone(), &spotify_email, &did, &token).await?; + Ok::<(), Error>(()) + }) { + Ok(_) => {} + Err(e) => { + println!( + "{} {}", + format!("[{}]", spotify_email).bright_green(), + e.to_string().bright_red() + ); + } + } + }); + } + } + + tokio::time::sleep(tokio::time::Duration::from_millis(1000)).await; + } + + Ok(()) +} diff --git a/crates/spotify/src/main.rs b/crates/spotify/src/main.rs index f5679442..4cc7ed85 100644 --- a/crates/spotify/src/main.rs +++ b/crates/spotify/src/main.rs @@ -7,28 +7,12 @@ use std::{ use anyhow::Error; use async_nats::connect; -use cache::Cache; -use crypto::decrypt_aes_256_ctr; use dotenv::dotenv; use owo_colors::OwoColorize; -use reqwest::Client; -use rocksky::{scrobble, update_library}; -use sqlx::{postgres::PgPoolOptions, Pool, Postgres}; +use rocksky_spotify::cache::Cache; +use rocksky_spotify::{find_spotify_user, find_spotify_users, watch_currently_playing}; +use sqlx::postgres::PgPoolOptions; use tokio_stream::StreamExt; -use types::{ - album_tracks::AlbumTracks, - currently_playing::{Album, Artist, CurrentlyPlaying}, - spotify_token::SpotifyTokenWithEmail, - token::AccessToken, -}; - -pub mod cache; -pub mod crypto; -pub mod rocksky; -pub mod token; -pub mod types; - -const BASE_URL: &str = "https://spotify-api.rocksky.app/v1"; #[tokio::main] async fn main() -> Result<(), Box> { @@ -226,672 +210,3 @@ async fn main() -> Result<(), Box> { Ok(()) } - -pub async fn refresh_token(token: &str) -> Result { - if env::var("SPOTIFY_CLIENT_ID").is_err() || env::var("SPOTIFY_CLIENT_SECRET").is_err() { - panic!("Please set SPOTIFY_CLIENT_ID and SPOTIFY_CLIENT_SECRET environment variables"); - } - - let client_id = env::var("SPOTIFY_CLIENT_ID")?; - let client_secret = env::var("SPOTIFY_CLIENT_SECRET")?; - - let client = Client::new(); - - let response = client - .post("https://accounts.spotify.com/api/token") - .basic_auth(&client_id, Some(client_secret)) - .form(&[ - ("grant_type", "refresh_token"), - ("refresh_token", token), - ("client_id", &client_id), - ]) - .send() - .await?; - let token = response.json::().await?; - Ok(token) -} - -pub async fn get_currently_playing( - cache: Cache, - user_id: &str, - token: &str, -) -> Result, Error> { - if let Ok(Some(data)) = cache.get(user_id) { - println!( - "{} {}", - format!("[{}]", user_id).bright_green(), - "Using cache".cyan() - ); - if data == "No content" { - return Ok(None); - } - let decoded_data = serde_json::from_str::(&data); - - if decoded_data.is_err() { - println!( - "{} {} {}", - format!("[{}]", user_id).bright_green(), - "Cache is invalid".red(), - data - ); - cache.setex(user_id, "No content", 10)?; - cache.del(&format!("{}:current", user_id))?; - return Ok(None); - } - - let data: CurrentlyPlaying = decoded_data.unwrap(); - // detect if the song has changed - let previous = cache.get(&format!("{}:previous", user_id)); - - if previous.is_err() { - println!( - "{} redis error: {}", - format!("[{}]", user_id).bright_green(), - previous.unwrap_err().to_string().bright_red() - ); - return Ok(None); - } - - let previous = previous.unwrap(); - - let changed = match previous { - Some(previous) => { - if serde_json::from_str::(&previous).is_err() { - println!( - "{} {} {}", - format!("[{}]", user_id).bright_green(), - "Previous cache is invalid", - previous - ); - return Ok(None); - } - - let previous: CurrentlyPlaying = serde_json::from_str(&previous)?; - if previous.item.is_none() && data.item.is_some() { - return Ok(Some((data, true))); - } - - if previous.item.is_some() && data.item.is_none() { - return Ok(Some((data, false))); - } - - if previous.item.is_none() && data.item.is_none() { - return Ok(Some((data, false))); - } - - let previous_item = previous.item.unwrap(); - let data_item = data.clone().item.unwrap(); - previous_item.id != data_item.id - && previous.progress_ms.unwrap_or(0) != data.progress_ms.unwrap_or(0) - } - _ => true, - }; - return Ok(Some((data, changed))); - } - - let token = refresh_token(token).await?; - let client = Client::new(); - let response = client - .get(format!("{}/me/player/currently-playing", BASE_URL)) - .bearer_auth(token.access_token) - .send() - .await?; - - let headers = response.headers().clone(); - let status = response.status().as_u16(); - let data = response.text().await?; - - if status == 429 { - println!( - "{} Too many requests, retry-after {}", - format!("[{}]", user_id).bright_green(), - headers - .get("retry-after") - .unwrap() - .to_str() - .unwrap() - .bright_green() - ); - return Ok(None); - } - - let previous = cache.get(&format!("{}:previous", user_id)); - if previous.is_err() { - println!( - "{} redis error: {}", - format!("[{}]", user_id).bright_green(), - previous.unwrap_err().to_string().bright_red() - ); - return Ok(None); - } - - let previous = previous.unwrap(); - - // check if status code is 204 - if status == 204 { - println!("No content"); - match cache.setex( - user_id, - "No content", - match previous.is_none() { - true => 30, - false => 10, - }, - ) { - Ok(_) => {} - Err(e) => { - println!( - "{} redis error: {}", - format!("[{}]", user_id).bright_green(), - e.to_string().bright_red() - ); - return Ok(None); - } - } - match cache.del(&format!("{}:current", user_id)) { - Ok(_) => {} - Err(e) => { - println!( - "{} redis error: {}", - format!("[{}]", user_id).bright_green(), - e.to_string().bright_red() - ); - return Ok(None); - } - } - return Ok(None); - } - - if serde_json::from_str::(&data).is_err() { - println!( - "{} {} {}", - format!("[{}]", user_id).bright_green(), - "Invalid data received".red(), - data - ); - match cache.setex(user_id, "No content", 10) { - Ok(_) => {} - Err(e) => { - println!( - "{} redis error: {}", - format!("[{}]", user_id).bright_green(), - e.to_string().bright_red() - ); - return Ok(None); - } - } - match cache.del(&format!("{}:current", user_id)) { - Ok(_) => {} - Err(e) => { - println!( - "{} redis error: {}", - format!("[{}]", user_id).bright_green(), - e.to_string().bright_red() - ); - return Ok(None); - } - } - return Ok(None); - } - - let data = serde_json::from_str::(&data)?; - - match cache.setex( - user_id, - &serde_json::to_string(&data)?, - match previous.is_none() { - true => 30, - false => 15, - }, - ) { - Ok(_) => {} - Err(e) => { - println!( - "{} redis error: {}", - format!("[{}]", user_id).bright_green(), - e.to_string().bright_red() - ); - return Ok(None); - } - } - match cache.del(&format!("{}:current", user_id)) { - Ok(_) => {} - Err(e) => { - println!( - "{} redis error: {}", - format!("[{}]", user_id).bright_green(), - e.to_string().bright_red() - ); - return Ok(None); - } - } - - // detect if the song has changed - let previous = cache.get(&format!("{}:previous", user_id)); - - if previous.is_err() { - println!( - "{} redis error: {}", - format!("[{}]", user_id).bright_green(), - previous.unwrap_err().to_string().bright_red() - ); - return Ok(None); - } - - let previous = previous.unwrap(); - let changed = match previous { - Some(previous) => { - if serde_json::from_str::(&previous).is_err() { - println!( - "{} {} {}", - format!("[{}]", user_id).bright_green(), - "Previous cache is invalid", - previous - ); - return Ok(None); - } - - let previous: CurrentlyPlaying = serde_json::from_str(&previous)?; - if previous.item.is_none() || data.item.is_none() { - return Ok(Some((data, false))); - } - - let previous_item = previous.item.unwrap(); - let data_item = data.clone().item.unwrap(); - - previous_item.id != data_item.id - && previous.progress_ms.unwrap_or(0) != data.progress_ms.unwrap_or(0) - } - _ => false, - }; - - // save as previous song - match cache.setex( - &format!("{}:previous", user_id), - &serde_json::to_string(&data)?, - 600, - ) { - Ok(_) => {} - Err(e) => { - println!( - "{} redis error: {}", - format!("[{}]", user_id).bright_green(), - e.to_string().bright_red() - ); - return Ok(None); - } - } - - Ok(Some((data, changed))) -} - -pub async fn get_artist( - cache: Cache, - artist_id: &str, - token: &str, -) -> Result, Error> { - if let Ok(Some(data)) = cache.get(artist_id) { - return Ok(Some(serde_json::from_str(&data)?)); - } - - let token = refresh_token(token).await?; - let client = Client::new(); - let response = client - .get(&format!("{}/artists/{}", BASE_URL, artist_id)) - .bearer_auth(token.access_token) - .send() - .await?; - - let headers = response.headers().clone(); - let data = response.text().await?; - - if data == "Too many requests" { - println!( - "> retry-after {}", - headers.get("retry-after").unwrap().to_str().unwrap() - ); - println!("> {} [get_artist]", data); - return Ok(None); - } - - match cache.setex(artist_id, &data, 20) { - Ok(_) => {} - Err(e) => { - println!( - "{} redis error: {}", - format!("[{}]", artist_id).bright_green(), - e.to_string().bright_red() - ); - return Ok(None); - } - } - - Ok(Some(serde_json::from_str(&data)?)) -} - -pub async fn get_album(cache: Cache, album_id: &str, token: &str) -> Result, Error> { - if let Ok(Some(data)) = cache.get(album_id) { - return Ok(Some(serde_json::from_str(&data)?)); - } - - let token = refresh_token(token).await?; - let client = Client::new(); - let response = client - .get(&format!("{}/albums/{}", BASE_URL, album_id)) - .bearer_auth(token.access_token) - .send() - .await?; - - let headers = response.headers().clone(); - let data = response.text().await?; - - if data == "Too many requests" { - println!( - "> retry-after {}", - headers.get("retry-after").unwrap().to_str().unwrap() - ); - println!("> {} [get_album]", data); - return Ok(None); - } - - match cache.setex(album_id, &data, 20) { - Ok(_) => {} - Err(e) => { - println!( - "{} redis error: {}", - format!("[{}]", album_id).bright_green(), - e.to_string().bright_red() - ); - return Ok(None); - } - } - - Ok(Some(serde_json::from_str(&data)?)) -} - -pub async fn get_album_tracks( - cache: Cache, - album_id: &str, - token: &str, -) -> Result { - if let Ok(Some(data)) = cache.get(&format!("{}:tracks", album_id)) { - return Ok(serde_json::from_str(&data)?); - } - - let token = refresh_token(token).await?; - let client = Client::new(); - let mut all_tracks = Vec::new(); - let mut offset = 0; - let limit = 50; - - loop { - let response = client - .get(&format!("{}/albums/{}/tracks", BASE_URL, album_id)) - .bearer_auth(&token.access_token) - .query(&[ - ("limit", &limit.to_string()), - ("offset", &offset.to_string()), - ]) - .send() - .await?; - - let headers = response.headers().clone(); - let data = response.text().await?; - if data == "Too many requests" { - println!( - "> retry-after {}", - headers.get("retry-after").unwrap().to_str().unwrap() - ); - println!("> {} [get_album_tracks]", data); - continue; - } - - let album_tracks: AlbumTracks = serde_json::from_str(&data)?; - - if album_tracks.items.is_empty() { - break; - } - - all_tracks.extend(album_tracks.items); - offset += limit; - } - - let all_tracks_json = serde_json::to_string(&all_tracks)?; - match cache.setex(&format!("{}:tracks", album_id), &all_tracks_json, 20) { - Ok(_) => {} - Err(e) => { - println!( - "{} redis error: {}", - format!("[{}]", album_id).bright_green(), - e.to_string().bright_red() - ); - } - } - - Ok(AlbumTracks { - items: all_tracks, - ..Default::default() - }) -} - -pub async fn find_spotify_users( - pool: &Pool, - offset: usize, - limit: usize, -) -> Result, Error> { - let results: Vec = sqlx::query_as( - r#" - SELECT * FROM spotify_tokens - LEFT JOIN spotify_accounts ON spotify_tokens.user_id = spotify_accounts.user_id - LEFT JOIN users ON spotify_accounts.user_id = users.xata_id - LIMIT $1 OFFSET $2 - "#, - ) - .bind(limit as i64) - .bind(offset as i64) - .fetch_all(pool) - .await?; - - let mut user_tokens = vec![]; - - for result in &results { - let token = decrypt_aes_256_ctr( - &result.refresh_token, - &hex::decode(env::var("SPOTIFY_ENCRYPTION_KEY")?)?, - )?; - user_tokens.push(( - result.email.clone(), - token, - result.did.clone(), - result.user_id.clone(), - )); - } - - Ok(user_tokens) -} - -pub async fn find_spotify_user( - pool: &Pool, - email: &str, -) -> Result, Error> { - let result: Vec = sqlx::query_as( - r#" - SELECT * FROM spotify_tokens - LEFT JOIN spotify_accounts ON spotify_tokens.user_id = spotify_accounts.user_id - LEFT JOIN users ON spotify_accounts.user_id = users.xata_id - WHERE spotify_accounts.email = $1 - "#, - ) - .bind(email) - .fetch_all(pool) - .await?; - - match result.first() { - Some(result) => { - let token = decrypt_aes_256_ctr( - &result.refresh_token, - &hex::decode(env::var("SPOTIFY_ENCRYPTION_KEY")?)?, - )?; - Ok(Some((result.email.clone(), token, result.did.clone()))) - } - None => Ok(None), - } -} - -pub async fn watch_currently_playing( - spotify_email: String, - token: String, - did: String, - stop_flag: Arc, - cache: Cache, -) -> Result<(), Error> { - println!( - "{} {}", - format!("[{}]", spotify_email).bright_green(), - "Checking currently playing".cyan() - ); - - let stop_flag_clone = stop_flag.clone(); - let spotify_email_clone = spotify_email.clone(); - let cache_clone = cache.clone(); - thread::spawn(move || { - loop { - if stop_flag_clone.load(std::sync::atomic::Ordering::Relaxed) { - println!( - "{} Stopping Thread", - format!("[{}]", spotify_email_clone).bright_green() - ); - break; - } - if let Ok(Some(cached)) = cache_clone.get(&format!("{}:current", spotify_email_clone)) { - if serde_json::from_str::(&cached).is_err() { - thread::sleep(std::time::Duration::from_millis(800)); - continue; - } - - let mut current_song = serde_json::from_str::(&cached)?; - - if let Some(item) = current_song.item.clone() { - if current_song.is_playing - && current_song.progress_ms.unwrap_or(0) < item.duration_ms.into() - { - current_song.progress_ms = - Some(current_song.progress_ms.unwrap_or(0) + 800); - match cache_clone.setex( - &format!("{}:current", spotify_email_clone), - &serde_json::to_string(¤t_song)?, - 16, - ) { - Ok(_) => {} - Err(e) => { - println!( - "{} redis error: {}", - format!("[{}]", spotify_email_clone).bright_green(), - e.to_string().bright_red() - ); - } - } - thread::sleep(std::time::Duration::from_millis(800)); - continue; - } - } - continue; - } - - if let Ok(Some(cached)) = cache_clone.get(&spotify_email_clone) { - if cached == "No content" { - thread::sleep(std::time::Duration::from_millis(800)); - continue; - } - match cache_clone.setex(&format!("{}:current", spotify_email_clone), &cached, 16) { - Ok(_) => {} - Err(e) => { - println!( - "{} redis error: {}", - format!("[{}]", spotify_email_clone).bright_green(), - e.to_string().bright_red() - ); - } - } - } - - thread::sleep(std::time::Duration::from_millis(800)); - } - Ok::<(), Error>(()) - }); - - loop { - if stop_flag.load(std::sync::atomic::Ordering::Relaxed) { - println!( - "{} Stopping Thread", - format!("[{}]", spotify_email).bright_green() - ); - break; - } - let spotify_email = spotify_email.clone(); - let token = token.clone(); - let did = did.clone(); - let cache = cache.clone(); - - let currently_playing = get_currently_playing(cache.clone(), &spotify_email, &token).await; - let currently_playing = match currently_playing { - Ok(currently_playing) => currently_playing, - Err(e) => { - println!( - "{} {}", - format!("[{}]", spotify_email).bright_green(), - e.to_string().bright_red() - ); - tokio::time::sleep(tokio::time::Duration::from_millis(1000)).await; - continue; - } - }; - - if let Some((data, changed)) = currently_playing { - if data.item.is_none() { - println!( - "{} {}", - format!("[{}]", spotify_email).bright_green(), - "No song playing".yellow() - ); - tokio::time::sleep(tokio::time::Duration::from_millis(1000)).await; - continue; - } - let data_item = data.item.unwrap(); - println!( - "{} {} is_playing: {} changed: {}", - format!("[{}]", spotify_email).bright_green(), - format!("{} - {}", data_item.name, data_item.artists[0].name).yellow(), - data.is_playing, - changed - ); - - if changed { - scrobble(cache.clone(), &spotify_email, &did, &token).await?; - - thread::spawn(move || { - let rt = tokio::runtime::Runtime::new().unwrap(); - match rt.block_on(async { - get_album_tracks(cache.clone(), &data_item.album.id, &token).await?; - get_album(cache.clone(), &data_item.album.id, &token).await?; - update_library(cache.clone(), &spotify_email, &did, &token).await?; - Ok::<(), Error>(()) - }) { - Ok(_) => {} - Err(e) => { - println!( - "{} {}", - format!("[{}]", spotify_email).bright_green(), - e.to_string().bright_red() - ); - } - } - }); - } - } - - tokio::time::sleep(tokio::time::Duration::from_millis(1000)).await; - } - - Ok(()) -} diff --git a/crates/storage/Cargo.toml b/crates/storage/Cargo.toml index 0a8b656d..44c01e53 100644 --- a/crates/storage/Cargo.toml +++ b/crates/storage/Cargo.toml @@ -1,5 +1,5 @@ [package] -name = "storage" +name = "rocksky-storage" version = "0.1.0" authors.workspace = true edition.workspace = true diff --git a/crates/storage/src/lib.rs b/crates/storage/src/lib.rs new file mode 100644 index 00000000..8b137891 --- /dev/null +++ b/crates/storage/src/lib.rs @@ -0,0 +1 @@ + diff --git a/crates/tracklist/Cargo.toml b/crates/tracklist/Cargo.toml new file mode 100644 index 00000000..4ff6d2cf --- /dev/null +++ b/crates/tracklist/Cargo.toml @@ -0,0 +1,24 @@ +[package] +name = "rocksky-tracklist" +version = "0.1.0" +authors.workspace = true +edition.workspace = true +license.workspace = true +repository.workspace = true + + +[dependencies] +redis = { version = "0.29.0", features = ["tokio-rustls-comp"] } +owo-colors = "4.1.0" +anyhow = "1.0.96" +async-nats = "0.39.0" +dotenv = "0.15.0" +serde = { version = "1.0.217", features = ["derive"] } +serde_json = "1.0.139" +tokio = { version = "1.43.0", features = ["full"] } +tokio-stream = { version = "0.1.17", features = ["full"] } +actix-web = "4.9.0" +polars = "0.46.0" +clap = "4.5.31" +rand = "0.9.2" +uuid = { version = "1.18.0", features = ["v4"] } diff --git a/crates/tracklist/src/handlers/mod.rs b/crates/tracklist/src/handlers/mod.rs new file mode 100644 index 00000000..2c3d2174 --- /dev/null +++ b/crates/tracklist/src/handlers/mod.rs @@ -0,0 +1,47 @@ +use crate::handlers::tracklist::*; +use actix_web::{web, HttpRequest, HttpResponse}; +use anyhow::Error; +use std::sync::Arc; + +pub mod tracklist; + +#[macro_export] +macro_rules! read_payload { + ($payload:expr) => {{ + let mut body = Vec::new(); + while let Some(chunk) = $payload.next().await { + // skip if None + match chunk { + Ok(bytes) => body.extend_from_slice(&bytes), + Err(err) => return Err(err.into()), + } + } + body + }}; +} + +pub async fn handle( + method: &str, + payload: &mut web::Payload, + req: &HttpRequest, + conn: Arc, +) -> Result { + match method { + "tracklist.addTrack" => add_track(payload, req, conn.clone()).await, + "tracklist.insertTrackAt" => insert_track_at(payload, req, conn.clone()).await, + "tracklist.removeTrackAt" => remove_track_at(payload, req, conn.clone()).await, + "tracklist.shuffleQueue" => shuffle_queue(payload, req, conn.clone()).await, + "tracklist.getQueue" => get_queue(payload, req, conn.clone()).await, + "tracklist.clearQueue" => clear_queue(payload, req, conn.clone()).await, + "tracklist.getQueueLength" => get_queue_length(payload, req, conn.clone()).await, + "tracklist.isQueueEmpty" => is_queue_empty(payload, req, conn.clone()).await, + "tracklist.setCurrentTrack" => set_current_track(payload, req, conn.clone()).await, + "tracklist.getCurrentTrack" => get_current_track(payload, req, conn.clone()).await, + "tracklist.clearCurrentTrack" => clear_current_track(payload, req, conn.clone()).await, + "tracklist.moveTrack" => move_track(payload, req, conn.clone()).await, + "tracklist.replaceQueue" => replace_queue(payload, req, conn.clone()).await, + "tracklist.getTrackAt" => get_track_at(payload, req, conn.clone()).await, + "tracklist.insertTracksAt" => insert_tracks_at(payload, req, conn.clone()).await, + _ => return Err(anyhow::anyhow!("Method not found")), + } +} diff --git a/crates/tracklist/src/handlers/tracklist.rs b/crates/tracklist/src/handlers/tracklist.rs new file mode 100644 index 00000000..fcfbeb1e --- /dev/null +++ b/crates/tracklist/src/handlers/tracklist.rs @@ -0,0 +1,205 @@ +use std::sync::Arc; + +use actix_web::{web, HttpRequest, HttpResponse}; +use anyhow::Error; +use serde_json::json; +use tokio_stream::StreamExt; + +use crate::{queue, read_payload, types::*}; + +pub async fn add_track( + payload: &mut web::Payload, + _req: &HttpRequest, + client: Arc, +) -> Result { + let body = read_payload!(payload); + let params = serde_json::from_slice::(&body)?; + + let new_queue = queue::add_track(&client, ¶ms.did, ¶ms.track_id).await?; + + Ok(HttpResponse::Ok().json(web::Json(json!(new_queue)))) +} + +pub async fn insert_track_at( + payload: &mut web::Payload, + _req: &HttpRequest, + client: Arc, +) -> Result { + let body = read_payload!(payload); + let params = serde_json::from_slice::(&body)?; + + let new_queue = + queue::insert_track_at(&client, ¶ms.did, params.index, ¶ms.track_id).await?; + + Ok(HttpResponse::Ok().json(web::Json(json!(new_queue)))) +} + +pub async fn remove_track_at( + payload: &mut web::Payload, + _req: &HttpRequest, + client: Arc, +) -> Result { + let body = read_payload!(payload); + let params = serde_json::from_slice::(&body)?; + + let new_queue = queue::remove_track_at(&client, ¶ms.did, params.index).await?; + + Ok(HttpResponse::Ok().json(web::Json(json!(new_queue)))) +} + +pub async fn shuffle_queue( + payload: &mut web::Payload, + _req: &HttpRequest, + client: Arc, +) -> Result { + let body = read_payload!(payload); + let params = serde_json::from_slice::(&body)?; + + let shuffled_queue = queue::shuffle_queue(&client, ¶ms.did).await?; + + Ok(HttpResponse::Ok().json(web::Json(json!(shuffled_queue)))) +} + +pub async fn get_queue( + payload: &mut web::Payload, + _req: &HttpRequest, + client: Arc, +) -> Result { + let body = read_payload!(payload); + let params = serde_json::from_slice::(&body)?; + + let tracks = queue::get_queue(&client, ¶ms.did).await?; + + Ok(HttpResponse::Ok().json(web::Json(tracks))) +} + +pub async fn clear_queue( + payload: &mut web::Payload, + _req: &HttpRequest, + client: Arc, +) -> Result { + let body = read_payload!(payload); + let params = serde_json::from_slice::(&body)?; + + queue::clear_queue(&client, ¶ms.did).await?; + + Ok(HttpResponse::Ok().json(web::Json(json!({})))) +} + +pub async fn get_queue_length( + payload: &mut web::Payload, + _req: &HttpRequest, + client: Arc, +) -> Result { + let body = read_payload!(payload); + let params = serde_json::from_slice::(&body)?; + + let length = queue::get_queue_length(&client, ¶ms.did).await?; + + Ok(HttpResponse::Ok().json(web::Json(json!({ "length": length })))) +} + +pub async fn is_queue_empty( + payload: &mut web::Payload, + _req: &HttpRequest, + client: Arc, +) -> Result { + let body = read_payload!(payload); + let params = serde_json::from_slice::(&body)?; + + let is_empty = queue::is_queue_empty(&client, ¶ms.did).await?; + + Ok(HttpResponse::Ok().json(web::Json(json!({ "is_empty": is_empty })))) +} + +pub async fn set_current_track( + payload: &mut web::Payload, + _req: &HttpRequest, + client: Arc, +) -> Result { + let body = read_payload!(payload); + let params = serde_json::from_slice::(&body)?; + + queue::set_current_track(&client, ¶ms.did, params.index).await?; + + Ok(HttpResponse::Ok().json(web::Json(json!({})))) +} + +pub async fn get_current_track( + payload: &mut web::Payload, + _req: &HttpRequest, + client: Arc, +) -> Result { + let body = read_payload!(payload); + let params = serde_json::from_slice::(&body)?; + + let current_track = queue::get_current_track(&client, ¶ms.did).await?; + + Ok(HttpResponse::Ok().json(web::Json(json!({ "current_track": current_track })))) +} + +pub async fn clear_current_track( + payload: &mut web::Payload, + _req: &HttpRequest, + client: Arc, +) -> Result { + let body = read_payload!(payload); + let params = serde_json::from_slice::(&body)?; + + queue::clear_current_track(&client, ¶ms.did).await?; + + Ok(HttpResponse::Ok().json(web::Json(json!({})))) +} + +pub async fn move_track( + payload: &mut web::Payload, + _req: &HttpRequest, + client: Arc, +) -> Result { + let body = read_payload!(payload); + let params = serde_json::from_slice::(&body)?; + + let new_queue = queue::move_track(&client, ¶ms.did, params.from, params.to).await?; + + Ok(HttpResponse::Ok().json(web::Json(json!(new_queue)))) +} + +pub async fn replace_queue( + payload: &mut web::Payload, + _req: &HttpRequest, + client: Arc, +) -> Result { + let body = read_payload!(payload); + let params = serde_json::from_slice::(&body)?; + + let new_queue = queue::replace_queue(&client, ¶ms.did, params.track_ids).await?; + + Ok(HttpResponse::Ok().json(web::Json(json!(new_queue)))) +} + +pub async fn get_track_at( + payload: &mut web::Payload, + _req: &HttpRequest, + client: Arc, +) -> Result { + let body = read_payload!(payload); + let params = serde_json::from_slice::(&body)?; + + let track_id = queue::get_track_at(&client, ¶ms.did, params.index).await?; + + Ok(HttpResponse::Ok().json(web::Json(json!({ "track_id": track_id })))) +} + +pub async fn insert_tracks_at( + payload: &mut web::Payload, + _req: &HttpRequest, + client: Arc, +) -> Result { + let body = read_payload!(payload); + let params = serde_json::from_slice::(&body)?; + + let new_queue = + queue::insert_tracks_at(&client, ¶ms.did, params.index, params.track_ids).await?; + + Ok(HttpResponse::Ok().json(web::Json(json!(new_queue)))) +} diff --git a/crates/tracklist/src/lib.rs b/crates/tracklist/src/lib.rs new file mode 100644 index 00000000..61ddcc46 --- /dev/null +++ b/crates/tracklist/src/lib.rs @@ -0,0 +1,11 @@ +pub mod handlers; +pub mod queue; +pub mod server; +pub mod types; + +use anyhow::Error; + +pub async fn run() -> Result<(), Error> { + server::run().await?; + Ok(()) +} diff --git a/crates/tracklist/src/queue.rs b/crates/tracklist/src/queue.rs new file mode 100644 index 00000000..9480d49a --- /dev/null +++ b/crates/tracklist/src/queue.rs @@ -0,0 +1,609 @@ +use anyhow::Error; +use rand::seq::SliceRandom; +use redis::AsyncCommands; + +pub async fn add_track( + client: &redis::Client, + did: &str, + track_id: &str, +) -> Result, Error> { + let mut conn = client.get_multiplexed_async_connection().await?; + + conn.rpush::<_, _, i32>(format!("user:{}:queue", did), track_id) + .await?; + + let queue: Vec = conn.lrange(format!("user:{}:queue", did), 0, -1).await?; + + Ok(queue) +} + +pub async fn insert_track_at( + client: &redis::Client, + did: &str, + position: usize, + track_id: &str, +) -> Result, Error> { + let mut conn = client.get_multiplexed_async_connection().await?; + + let queue: Vec = conn.lrange(format!("user:{}:queue", did), 0, -1).await?; + + let mut new_queue = queue.clone(); + if position >= new_queue.len() { + new_queue.push(track_id.to_string()); + } else { + new_queue.insert(position, track_id.to_string()); + } + + let mut pipeline = redis::pipe(); + pipeline + .atomic() + .del(format!("user:{}:queue", did)) + .rpush(format!("user:{}:queue", did), new_queue.clone()) + .query_async::<()>(&mut conn) + .await?; + + Ok(new_queue) +} + +pub async fn remove_track_at( + client: &redis::Client, + did: &str, + position: usize, +) -> Result, Error> { + let mut conn = client.get_multiplexed_async_connection().await?; + + let queue: Vec = conn.lrange(format!("user:{}:queue", did), 0, -1).await?; + if position < queue.len() { + let _: i32 = conn + .lrem::<_, _, i32>(format!("user:{}:queue", did), 1, &queue[position]) + .await?; + } + + let new_queue: Vec = conn.lrange(format!("user:{}:queue", did), 0, -1).await?; + Ok(new_queue) +} + +pub async fn shuffle_queue(client: &redis::Client, did: &str) -> Result, Error> { + let mut conn = client.get_multiplexed_async_connection().await?; + + let mut queue: Vec = conn.lrange(format!("user:{}:queue", did), 0, -1).await?; + let old_queue = queue.clone(); + + loop { + let mut rng = rand::rng(); + queue.shuffle(&mut rng); + if queue != old_queue { + break; + } + } + + redis::pipe() + .atomic() + .del(format!("user:{}:queue", did)) + .rpush(format!("user:{}:queue", did), queue.clone()) + .query_async::<()>(&mut conn) + .await?; + + Ok(queue) +} + +pub async fn get_queue(client: &redis::Client, did: &str) -> Result, Error> { + let mut conn = client.get_multiplexed_async_connection().await?; + + let queue: Vec = conn.lrange(format!("user:{}:queue", did), 0, -1).await?; + Ok(queue) +} + +pub async fn clear_queue(client: &redis::Client, did: &str) -> Result<(), Error> { + let mut conn = client.get_multiplexed_async_connection().await?; + + redis::pipe() + .atomic() + .del(format!("user:{}:queue", did)) + .query_async::<()>(&mut conn) + .await?; + + Ok(()) +} + +pub async fn get_queue_length(client: &redis::Client, did: &str) -> Result { + let mut conn = client.get_multiplexed_async_connection().await?; + + let length: usize = conn.llen(format!("user:{}:queue", did)).await?; + Ok(length) +} + +pub async fn is_queue_empty(client: &redis::Client, did: &str) -> Result { + let length = get_queue_length(client, did).await?; + Ok(length == 0) +} + +pub async fn set_current_track( + client: &redis::Client, + did: &str, + position: usize, +) -> Result<(), Error> { + let mut conn = client.get_multiplexed_async_connection().await?; + + conn.set::<_, _, ()>(format!("user:{}:current_track", did), position) + .await?; + + Ok(()) +} + +pub async fn get_current_track(client: &redis::Client, did: &str) -> Result, Error> { + let mut conn = client.get_multiplexed_async_connection().await?; + + let position: Option = conn + .get::<_, Option>(format!("user:{}:current_track", did)) + .await?; + + Ok(position) +} + +pub async fn clear_current_track(client: &redis::Client, did: &str) -> Result<(), Error> { + let mut conn = client.get_multiplexed_async_connection().await?; + + conn.del::<_, ()>(format!("user:{}:current_track", did)) + .await?; + + Ok(()) +} + +pub async fn move_track( + client: &redis::Client, + did: &str, + from: usize, + to: usize, +) -> Result, Error> { + let mut conn = client.get_multiplexed_async_connection().await?; + + let queue: Vec = conn.lrange(format!("user:{}:queue", did), 0, -1).await?; + if from >= queue.len() || to >= queue.len() { + return Ok(queue); + } + + let mut new_queue = queue.clone(); + let track = new_queue.remove(from); + new_queue.insert(to, track); + + redis::pipe() + .atomic() + .del(format!("user:{}:queue", did)) + .rpush(format!("user:{}:queue", did), new_queue.clone()) + .query_async::<()>(&mut conn) + .await?; + + Ok(new_queue) +} + +pub async fn replace_queue( + client: &redis::Client, + did: &str, + new_queue: Vec, +) -> Result, Error> { + let mut conn = client.get_multiplexed_async_connection().await?; + + redis::pipe() + .atomic() + .del(format!("user:{}:queue", did)) + .rpush(format!("user:{}:queue", did), new_queue.clone()) + .query_async::<()>(&mut conn) + .await?; + + Ok(new_queue) +} + +pub async fn get_track_at( + client: &redis::Client, + did: &str, + position: usize, +) -> Result, Error> { + let mut conn = client.get_multiplexed_async_connection().await?; + + let track: Option = conn + .lindex::<_, Option>(format!("user:{}:queue", did), position as isize) + .await?; + + Ok(track) +} + +pub async fn insert_tracks_at( + client: &redis::Client, + did: &str, + position: usize, + track_ids: Vec, +) -> Result, Error> { + let mut conn = client.get_multiplexed_async_connection().await?; + + let queue: Vec = conn.lrange(format!("user:{}:queue", did), 0, -1).await?; + + let mut new_queue = queue.clone(); + if position >= new_queue.len() { + new_queue.extend(track_ids); + } else { + for (i, track_id) in track_ids.into_iter().enumerate() { + new_queue.insert(position + i, track_id); + } + } + + let mut pipeline = redis::pipe(); + pipeline + .atomic() + .del(format!("user:{}:queue", did)) + .rpush(format!("user:{}:queue", did), new_queue.clone()) + .query_async::<()>(&mut conn) + .await?; + + Ok(new_queue) +} + +#[cfg(test)] +mod tests { + use super::*; + use anyhow::Error; + use redis::AsyncCommands; + use uuid::Uuid; + + async fn setup_redis() -> redis::Client { + redis::Client::open("redis://localhost:6379/").expect("Failed to create Redis client") + } + + async fn cleanup(client: &redis::Client, did: &str) -> Result<(), Error> { + let mut conn = client.get_multiplexed_async_connection().await?; + conn.del::<_, ()>(format!("user:{}:queue", did)).await?; + Ok(()) + } + + #[tokio::test] + async fn test_add_track() -> Result<(), Error> { + let client = setup_redis().await; + let did = Uuid::new_v4().to_string(); + let track_id = "track:67890"; + + // Add a track + add_track(&client, &did, track_id).await?; + let queue = get_queue(&client, &did).await?; + assert_eq!(queue, vec![track_id]); + + // Add another track + let track_id2 = "track:67891"; + add_track(&client, &did, track_id2).await?; + let queue = get_queue(&client, &did).await?; + assert_eq!(queue, vec![track_id, track_id2]); + + // Cleanup + cleanup(&client, &did).await?; + Ok(()) + } + + #[tokio::test] + async fn test_insert_track_at() -> Result<(), Error> { + let client = setup_redis().await; + let did = Uuid::new_v4().to_string(); + let track_ids = vec!["track:67890", "track:67891", "track:67892"]; + + for &track_id in &track_ids { + add_track(&client, &did, track_id).await?; + } + + let new_track = "track:67893"; + insert_track_at(&client, &did, 1, new_track).await?; + let queue: Vec = get_queue(&client, &did).await?; + assert_eq!( + queue, + vec!["track:67890", "track:67893", "track:67891", "track:67892"] + ); + + let end_track = "track:67894"; + insert_track_at(&client, &did, 10, end_track).await?; + let queue = get_queue(&client, &did).await?; + assert_eq!( + queue, + vec![ + "track:67890", + "track:67893", + "track:67891", + "track:67892", + "track:67894" + ] + ); + + let new_did = Uuid::new_v4().to_string(); + insert_track_at(&client, &new_did, 0, "track:67895").await?; + let queue = get_queue(&client, &new_did).await?; + assert_eq!(queue, vec!["track:67895"]); + + cleanup(&client, &did).await?; + cleanup(&client, &new_did).await?; + Ok(()) + } + + #[tokio::test] + async fn test_remove_track_at() -> Result<(), Error> { + let client = setup_redis().await; + let did = Uuid::new_v4().to_string(); + let track_ids = vec!["track:67890", "track:67891", "track:67892"]; + + for &track_id in &track_ids { + add_track(&client, &did, track_id).await?; + } + + remove_track_at(&client, &did, 1).await?; + let queue = get_queue(&client, &did).await?; + assert_eq!(queue, vec!["track:67890", "track:67892"]); + + remove_track_at(&client, &did, 0).await?; + let queue = get_queue(&client, &did).await?; + assert_eq!(queue, vec!["track:67892"]); + + remove_track_at(&client, &did, 5).await?; + let queue = get_queue(&client, &did).await?; + assert_eq!(queue, vec!["track:67892"]); + + let new_did = Uuid::new_v4().to_string(); + remove_track_at(&client, &new_did, 0).await?; + let queue = get_queue(&client, &new_did).await?; + assert_eq!(queue, Vec::::new()); + + cleanup(&client, &did).await?; + cleanup(&client, &new_did).await?; + Ok(()) + } + + #[tokio::test] + async fn test_shuffle_queue() -> Result<(), Error> { + let client = setup_redis().await; + let did = Uuid::new_v4().to_string(); + let track_ids = vec!["track:67890", "track:67891", "track:67892"]; + + for &track_id in &track_ids { + add_track(&client, &did, track_id).await?; + } + + shuffle_queue(&client, &did).await?; + let queue = get_queue(&client, &did).await?; + assert_eq!(queue.len(), track_ids.len()); + assert!(track_ids.iter().all(|id| queue.contains(&id.to_string()))); + + cleanup(&client, &did).await?; + Ok(()) + } + + #[tokio::test] + async fn test_get_queue() -> Result<(), Error> { + let client = setup_redis().await; + let did = Uuid::new_v4().to_string(); + let track_ids = vec!["track:67890", "track:67891"]; + + let queue = get_queue(&client, &did).await?; + assert_eq!(queue, Vec::::new()); + + for &track_id in &track_ids { + add_track(&client, &did, track_id).await?; + } + + let queue = get_queue(&client, &did).await?; + assert_eq!(queue, track_ids); + + cleanup(&client, &did).await?; + Ok(()) + } + + #[tokio::test] + async fn test_clear_queue() -> Result<(), Error> { + let client = setup_redis().await; + let did = Uuid::new_v4().to_string(); + let track_ids = vec!["track:67890", "track:67891"]; + + for &track_id in &track_ids { + add_track(&client, &did, track_id).await?; + } + + clear_queue(&client, &did).await?; + let queue = get_queue(&client, &did).await?; + assert_eq!(queue, Vec::::new()); + + clear_queue(&client, &did).await?; + let queue = get_queue(&client, &did).await?; + assert_eq!(queue, Vec::::new()); + + cleanup(&client, &did).await?; + Ok(()) + } + + #[tokio::test] + async fn test_queue_length_and_empty() -> Result<(), Error> { + let client = setup_redis().await; + let did = Uuid::new_v4().to_string(); + let track_ids = vec!["track:67890", "track:67891"]; + + let length = get_queue_length(&client, &did).await?; + assert_eq!(length, 0); + let is_empty = is_queue_empty(&client, &did).await?; + assert!(is_empty); + + for &track_id in &track_ids { + add_track(&client, &did, track_id).await?; + } + + let length = get_queue_length(&client, &did).await?; + assert_eq!(length, track_ids.len()); + let is_empty = is_queue_empty(&client, &did).await?; + assert!(!is_empty); + + cleanup(&client, &did).await?; + Ok(()) + } + + #[tokio::test] + async fn test_current_track() -> Result<(), Error> { + let client = setup_redis().await; + let did = Uuid::new_v4().to_string(); + let track_ids = vec!["track:67890", "track:67891"]; + for &track_id in &track_ids { + add_track(&client, &did, track_id).await?; + } + let current = get_current_track(&client, &did).await?; + assert_eq!(current, None); + set_current_track(&client, &did, 1).await?; + let current = get_current_track(&client, &did).await?; + assert_eq!(current, Some(1)); + clear_current_track(&client, &did).await?; + let current = get_current_track(&client, &did).await?; + assert_eq!(current, None); + cleanup(&client, &did).await?; + Ok(()) + } + + #[tokio::test] + async fn test_move_track() -> Result<(), Error> { + let client = setup_redis().await; + let did = Uuid::new_v4().to_string(); + let track_ids = vec!["track:67890", "track:67891", "track:67892"]; + + for &track_id in &track_ids { + add_track(&client, &did, track_id).await?; + } + + move_track(&client, &did, 0, 2).await?; + let queue = get_queue(&client, &did).await?; + assert_eq!(queue, vec!["track:67891", "track:67892", "track:67890"]); + move_track(&client, &did, 2, 0).await?; + + let queue = get_queue(&client, &did).await?; + assert_eq!(queue, vec!["track:67890", "track:67891", "track:67892"]); + move_track(&client, &did, 1, 1).await?; + + let queue = get_queue(&client, &did).await?; + assert_eq!(queue, vec!["track:67890", "track:67891", "track:67892"]); + move_track(&client, &did, 5, 0).await?; + + let queue = get_queue(&client, &did).await?; + assert_eq!(queue, vec!["track:67890", "track:67891", "track:67892"]); + + let new_did = Uuid::new_v4().to_string(); + move_track(&client, &new_did, 0, 1).await?; + + let queue = get_queue(&client, &new_did).await?; + assert_eq!(queue, Vec::::new()); + + cleanup(&client, &did).await?; + cleanup(&client, &new_did).await?; + + Ok(()) + } + + #[tokio::test] + async fn test_replace_queue() -> Result<(), Error> { + let client = setup_redis().await; + let did = Uuid::new_v4().to_string(); + let initial_tracks = vec!["track:67890", "track:67891"]; + + for &track_id in &initial_tracks { + add_track(&client, &did, track_id).await?; + } + + let new_queue = vec![ + "track:67892".to_string(), + "track:67893".to_string(), + "track:67894".to_string(), + ]; + + replace_queue(&client, &did, new_queue.clone()).await?; + let queue = get_queue(&client, &did).await?; + + assert_eq!(queue, new_queue); + cleanup(&client, &did).await?; + Ok(()) + } + + #[tokio::test] + async fn test_get_track_at() -> Result<(), Error> { + let client = setup_redis().await; + let did = Uuid::new_v4().to_string(); + let track_ids = vec!["track:67890", "track:67891", "track:67892"]; + + for &track_id in &track_ids { + add_track(&client, &did, track_id).await?; + } + + let track = get_track_at(&client, &did, 1).await?; + assert_eq!(track, Some("track:67891".to_string())); + + let track = get_track_at(&client, &did, 5).await?; + assert_eq!(track, None); + + let new_did = Uuid::new_v4().to_string(); + let track = get_track_at(&client, &new_did, 0).await?; + assert_eq!(track, None); + + cleanup(&client, &did).await?; + cleanup(&client, &new_did).await?; + + Ok(()) + } + + #[tokio::test] + async fn test_insert_tracks_at() -> Result<(), Error> { + let client = setup_redis().await; + let did = Uuid::new_v4().to_string(); + let initial_tracks = vec!["track:67890", "track:67891"]; + + for &track_id in &initial_tracks { + add_track(&client, &did, track_id).await?; + } + + let new_tracks = vec!["track:67892".to_string(), "track:67893".to_string()]; + insert_tracks_at(&client, &did, 1, new_tracks.clone()).await?; + let queue = get_queue(&client, &did).await?; + + assert_eq!( + queue, + vec!["track:67890", "track:67892", "track:67893", "track:67891"] + ); + + let end_tracks = vec!["track:67894".to_string()]; + insert_tracks_at(&client, &did, 10, end_tracks.clone()).await?; + let queue = get_queue(&client, &did).await?; + assert_eq!( + queue, + vec![ + "track:67890", + "track:67892", + "track:67893", + "track:67891", + "track:67894" + ] + ); + let new_did = Uuid::new_v4().to_string(); + let new_tracks = vec!["track:67895".to_string(), "track:67896".to_string()]; + insert_tracks_at(&client, &new_did, 0, new_tracks.clone()).await?; + + let queue = get_queue(&client, &new_did).await?; + assert_eq!(queue, new_tracks); + + cleanup(&client, &did).await?; + cleanup(&client, &new_did).await?; + Ok(()) + } + + #[tokio::test] + async fn test_concurrent_operations() -> Result<(), Error> { + let client = setup_redis().await; + let did = Uuid::new_v4().to_string(); + let track_ids = vec!["track:67890", "track:67891", "track:67892"]; + + let add_task = add_track(&client, &did, track_ids[0]); + let insert_task = insert_track_at(&client, &did, 0, track_ids[1]); + let remove_task = remove_track_at(&client, &did, 0); + tokio::try_join!(add_task, insert_task, remove_task)?; + + let queue = get_queue(&client, &did).await?; + assert!(queue.len() <= 2); + assert!(track_ids.iter().any(|id| queue.contains(&id.to_string()))); + + cleanup(&client, &did).await?; + Ok(()) + } +} diff --git a/crates/tracklist/src/server.rs b/crates/tracklist/src/server.rs new file mode 100644 index 00000000..1c192033 --- /dev/null +++ b/crates/tracklist/src/server.rs @@ -0,0 +1,60 @@ +use std::{env, sync::Arc}; + +use actix_web::{ + get, post, + web::{self, Data}, + App, HttpRequest, HttpResponse, HttpServer, Responder, +}; +use anyhow::Error; +use owo_colors::OwoColorize; +use serde_json::json; + +use crate::handlers::handle; + +#[get("/")] +async fn index(_req: HttpRequest) -> HttpResponse { + HttpResponse::Ok().json(json!({ + "server": "Rocksky Tracklist Server", + "version": "0.1.0", + })) +} + +#[post("/{method}")] +async fn call_method( + data: web::Data>, + mut payload: web::Payload, + req: HttpRequest, +) -> Result { + let method = req.match_info().get("method").unwrap_or("unknown"); + println!("Method: {}", method.bright_green()); + + let conn = data.get_ref().clone(); + handle(method, &mut payload, &req, conn) + .await + .map_err(actix_web::error::ErrorInternalServerError) +} + +pub async fn run() -> Result<(), Error> { + let host = env::var("TRACKLIST_HOST").unwrap_or_else(|_| "127.0.0.1".to_string()); + let port = env::var("TRACKLIST_PORT").unwrap_or_else(|_| "7884".to_string()); + let addr = format!("{}:{}", host, port); + + let url = format!("http://{}", addr); + println!("Listening on {}", url.bright_green()); + + let client = redis::Client::open(env::var("REDIS_URL").unwrap_or("redis://127.0.0.1".into()))?; + let conn = Arc::new(client); + + HttpServer::new(move || { + App::new() + .app_data(Data::new(conn.clone())) + .service(index) + .service(call_method) + }) + .bind(&addr)? + .run() + .await + .map_err(Error::new)?; + + Ok(()) +} diff --git a/crates/tracklist/src/types.rs b/crates/tracklist/src/types.rs new file mode 100644 index 00000000..97b6bb33 --- /dev/null +++ b/crates/tracklist/src/types.rs @@ -0,0 +1,87 @@ +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Serialize, Deserialize)] +pub struct AddTrackParams { + pub did: String, + pub track_id: String, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct InsertTrackAtParams { + pub did: String, + pub track_id: String, + pub index: usize, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct RemoveTrackAtParams { + pub did: String, + pub index: usize, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct ShuffleQueueParams { + pub did: String, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct GetQueueParams { + pub did: String, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct ClearQueueParams { + pub did: String, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct GetQueueLengthParams { + pub did: String, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct IsQueueEmptyParams { + pub did: String, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct SetCurrentTrackParams { + pub did: String, + pub index: usize, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct GetCurrentTrackParams { + pub did: String, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct ClearCurrentTrackParams { + pub did: String, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct MoveTrackParams { + pub did: String, + pub from: usize, + pub to: usize, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct ReplaceQueueParams { + pub did: String, + pub track_ids: Vec, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct GetTrackAtParams { + pub did: String, + pub index: usize, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct InsertTracksAtParams { + pub did: String, + pub track_ids: Vec, + pub index: usize, +} diff --git a/crates/webscrobbler/Cargo.toml b/crates/webscrobbler/Cargo.toml index c694a4f5..d7a0c5ac 100644 --- a/crates/webscrobbler/Cargo.toml +++ b/crates/webscrobbler/Cargo.toml @@ -1,5 +1,5 @@ [package] -name = "webscrobbler" +name = "rocksky-webscrobbler" version = "0.1.0" authors.workspace = true edition.workspace = true diff --git a/crates/webscrobbler/src/consts.rs b/crates/webscrobbler/src/consts.rs new file mode 100644 index 00000000..4338326f --- /dev/null +++ b/crates/webscrobbler/src/consts.rs @@ -0,0 +1,10 @@ +pub const BANNER: &str = r#" + _ __ __ _____ __ __ __ + | | / /__ / /_ / ___/______________ / /_ / /_ / /__ _____ + | | /| / / _ \/ __ \\__ \/ ___/ ___/ __ \/ __ \/ __ \/ / _ \/ ___/ + | |/ |/ / __/ /_/ /__/ / /__/ / / /_/ / /_/ / /_/ / / __/ / + |__/|__/\___/_.___/____/\___/_/ \____/_.___/_.___/_/\___/_/ + + + This is the Rocksky WebScrobbler Webhook API compatible with webscrobbler extension. +"#; diff --git a/crates/webscrobbler/src/handlers.rs b/crates/webscrobbler/src/handlers.rs index 224c2670..ee548fad 100644 --- a/crates/webscrobbler/src/handlers.rs +++ b/crates/webscrobbler/src/handlers.rs @@ -1,4 +1,4 @@ -use crate::{cache::Cache, repo, scrobbler::scrobble, types::ScrobbleRequest, BANNER}; +use crate::{cache::Cache, consts::BANNER, repo, scrobbler::scrobble, types::ScrobbleRequest}; use actix_web::{get, post, web, HttpRequest, HttpResponse, Responder}; use owo_colors::OwoColorize; use sqlx::{Pool, Postgres}; diff --git a/crates/webscrobbler/src/lib.rs b/crates/webscrobbler/src/lib.rs new file mode 100644 index 00000000..43f9cc3a --- /dev/null +++ b/crates/webscrobbler/src/lib.rs @@ -0,0 +1,78 @@ +use std::{env, sync::Arc, time::Duration}; + +use actix_limitation::{Limiter, RateLimiter}; +use actix_session::SessionExt; +use actix_web::{ + dev::ServiceRequest, + web::{self, Data}, + App, HttpServer, +}; +use anyhow::Error; +use owo_colors::OwoColorize; +use sqlx::postgres::PgPoolOptions; + +use crate::{cache::Cache, consts::BANNER}; + +pub mod auth; +pub mod cache; +pub mod consts; +pub mod crypto; +pub mod handlers; +pub mod musicbrainz; +pub mod repo; +pub mod rocksky; +pub mod scrobbler; +pub mod spotify; +pub mod types; +pub mod xata; + +pub async fn start_server() -> Result<(), Error> { + println!("{}", BANNER.magenta()); + + let cache = Cache::new()?; + + let pool = PgPoolOptions::new() + .max_connections(5) + .connect(&env::var("XATA_POSTGRES_URL")?) + .await?; + + let conn = Arc::new(pool); + + let host = env::var("WEBSCROBBLER_HOST").unwrap_or_else(|_| "127.0.0.1".to_string()); + let port = env::var("WEBSCROBBLER_PORT") + .unwrap_or_else(|_| "7883".to_string()) + .parse::() + .unwrap_or(7883); + + println!( + "Starting WebScrobbler Webhook @ {}", + format!("{}:{}", host, port).green() + ); + + let limiter = web::Data::new( + Limiter::builder("redis://127.0.0.1") + .key_by(|req: &ServiceRequest| { + req.get_session() + .get(&"session-id") + .unwrap_or_else(|_| req.cookie(&"rate-api-id").map(|c| c.to_string())) + }) + .limit(100) + .period(Duration::from_secs(60)) // 60 minutes + .build() + .unwrap(), + ); + + HttpServer::new(move || { + App::new() + .wrap(RateLimiter::default()) + .app_data(limiter.clone()) + .app_data(Data::new(conn.clone())) + .app_data(Data::new(cache.clone())) + .service(handlers::index) + .service(handlers::handle_scrobble) + }) + .bind((host, port))? + .run() + .await?; + Ok(()) +} diff --git a/crates/webscrobbler/src/main.rs b/crates/webscrobbler/src/main.rs index 93cddc6d..7c0f3f14 100644 --- a/crates/webscrobbler/src/main.rs +++ b/crates/webscrobbler/src/main.rs @@ -1,92 +1,12 @@ -use std::{env, sync::Arc, time::Duration}; - -use actix_limitation::{Limiter, RateLimiter}; -use actix_session::SessionExt as _; -use actix_web::{ - dev::ServiceRequest, - web::{self, Data}, - App, HttpServer, -}; use anyhow::Error; -use cache::Cache; use dotenv::dotenv; -use owo_colors::OwoColorize; -use sqlx::postgres::PgPoolOptions; - -pub mod auth; -pub mod cache; -pub mod crypto; -pub mod handlers; -pub mod musicbrainz; -pub mod repo; -pub mod rocksky; -pub mod scrobbler; -pub mod spotify; -pub mod types; -pub mod xata; - -pub const BANNER: &str = r#" - _ __ __ _____ __ __ __ - | | / /__ / /_ / ___/______________ / /_ / /_ / /__ _____ - | | /| / / _ \/ __ \\__ \/ ___/ ___/ __ \/ __ \/ __ \/ / _ \/ ___/ - | |/ |/ / __/ /_/ /__/ / /__/ / / /_/ / /_/ / /_/ / / __/ / - |__/|__/\___/_.___/____/\___/_/ \____/_.___/_.___/_/\___/_/ - - - This is the Rocksky WebScrobbler Webhook API compatible with webscrobbler extension. -"#; +use rocksky_webscrobbler::start_server; #[tokio::main] async fn main() -> Result<(), Error> { dotenv().ok(); - println!("{}", BANNER.magenta()); - - let cache = Cache::new()?; - - let pool = PgPoolOptions::new() - .max_connections(5) - .connect(&env::var("XATA_POSTGRES_URL")?) - .await?; - - let conn = Arc::new(pool); - - let host = env::var("WEBSCROBBLER_HOST").unwrap_or_else(|_| "127.0.0.1".to_string()); - let port = env::var("WEBSCROBBLER_PORT") - .unwrap_or_else(|_| "7883".to_string()) - .parse::() - .unwrap_or(7883); - - println!( - "Starting WebScrobbler Webhook @ {}", - format!("{}:{}", host, port).green() - ); - - let limiter = web::Data::new( - Limiter::builder("redis://127.0.0.1") - .key_by(|req: &ServiceRequest| { - req.get_session() - .get(&"session-id") - .unwrap_or_else(|_| req.cookie(&"rate-api-id").map(|c| c.to_string())) - }) - .limit(100) - .period(Duration::from_secs(60)) // 60 minutes - .build() - .unwrap(), - ); - - HttpServer::new(move || { - App::new() - .wrap(RateLimiter::default()) - .app_data(limiter.clone()) - .app_data(Data::new(conn.clone())) - .app_data(Data::new(cache.clone())) - .service(handlers::index) - .service(handlers::handle_scrobble) - }) - .bind((host, port))? - .run() - .await?; + start_server().await?; Ok(()) } -- 2.51.2