diff --git a/Cargo.lock b/Cargo.lock index ec7525c3..02c1fe3f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1102,6 +1102,17 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" +[[package]] +name = "chacha20" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3613f74bd2eac03dad61bd53dbe620703d4371614fe0bc3b9f04dd36fe4e818" +dependencies = [ + "cfg-if", + "cipher", + "cpufeatures", +] + [[package]] name = "chrono" version = "0.4.39" @@ -1487,6 +1498,7 @@ dependencies = [ "fiat-crypto", "rustc_version", "subtle", + "zeroize", ] [[package]] @@ -1662,6 +1674,26 @@ version = "0.15.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b" +[[package]] +name = "dryoc" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e73e0fee365832cd9b9a53ea62f944cc0d7a4c71f2b9c96a28fc74749517afa" +dependencies = [ + "bitflags 2.9.4", + "chacha20", + "curve25519-dalek", + "generic-array", + "lazy_static", + "libc", + "rand_core 0.6.4", + "salsa20", + "sha2", + "subtle", + "winapi", + "zeroize", +] + [[package]] name = "duckdb" version = "1.2.2" @@ -3157,6 +3189,15 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ffbee8634e0d45d258acb448e7eaab3fce7a0a467395d4d9f228e3c1f01fb2e4" +[[package]] +name = "matchers" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9" +dependencies = [ + "regex-automata", +] + [[package]] name = "maybe-async" version = "0.2.10" @@ -5001,6 +5042,32 @@ dependencies = [ "url", ] +[[package]] +name = "rocksky-mirror" +version = "0.1.0" +dependencies = [ + "anyhow", + "async-nats", + "base64 0.22.1", + "chrono", + "dotenv", + "dryoc", + "futures-util", + "hex", + "jsonwebtoken", + "owo-colors", + "reqwest", + "serde", + "serde_json", + "sqlx", + "tokio", + "tokio-stream", + "tokio-tungstenite", + "tokio-util", + "tracing", + "tracing-subscriber", +] + [[package]] name = "rocksky-navidrome" version = "0.1.0" @@ -5207,6 +5274,7 @@ dependencies = [ "rocksky-dropbox", "rocksky-googledrive", "rocksky-jetstream", + "rocksky-mirror", "rocksky-navidrome", "rocksky-pgpull", "rocksky-playlists", @@ -6811,6 +6879,7 @@ dependencies = [ "futures-core", "futures-io", "futures-sink", + "futures-util", "pin-project-lite", "tokio", ] @@ -7023,10 +7092,14 @@ version = "0.3.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2054a14f5307d601f88daf0553e1cbf472acc4f2c51afab632431cdcd72124d5" dependencies = [ + "matchers", "nu-ansi-term", + "once_cell", + "regex-automata", "sharded-slab", "smallvec", "thread_local", + "tracing", "tracing-core", "tracing-log", ] @@ -8032,6 +8105,20 @@ name = "zeroize" version = "1.8.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ced3678a2879b30306d323f4542626697a464a97c0a07c9aebf7ebca65cd4dde" +dependencies = [ + "zeroize_derive", +] + +[[package]] +name = "zeroize_derive" +version = "1.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85a5b4158499876c763cb03bc4e49185d3cccbabb15b33c627f7884f43db852e" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.106", +] [[package]] name = "zerotrie" diff --git a/apps/api/drizzle/0010_mirror_sources.sql b/apps/api/drizzle/0010_mirror_sources.sql new file mode 100644 index 00000000..7b8b87a7 --- /dev/null +++ b/apps/api/drizzle/0010_mirror_sources.sql @@ -0,0 +1,19 @@ +CREATE TABLE "mirror_sources" ( + "xata_id" text PRIMARY KEY DEFAULT xata_id() NOT NULL, + "user_id" text NOT NULL, + "provider" text NOT NULL, + "enabled" boolean DEFAULT false NOT NULL, + "external_username" text, + "encrypted_api_key" text, + "last_polled_at" timestamp, + "last_scrobble_seen_at" timestamp, + "xata_createdat" timestamp DEFAULT now() NOT NULL, + "xata_updatedat" timestamp DEFAULT now() NOT NULL, + "xata_version" integer +); +--> statement-breakpoint +ALTER TABLE "mirror_sources" ADD CONSTRAINT "mirror_sources_user_id_users_xata_id_fk" FOREIGN KEY ("user_id") REFERENCES "public"."users"("xata_id") ON DELETE no action ON UPDATE no action; +--> statement-breakpoint +CREATE UNIQUE INDEX "mirror_sources_user_provider_idx" ON "mirror_sources" USING btree ("user_id","provider"); +--> statement-breakpoint +CREATE INDEX "mirror_sources_enabled_provider_idx" ON "mirror_sources" USING btree ("enabled","provider"); diff --git a/apps/api/drizzle/meta/_journal.json b/apps/api/drizzle/meta/_journal.json index a4819331..76f0da6a 100644 --- a/apps/api/drizzle/meta/_journal.json +++ b/apps/api/drizzle/meta/_journal.json @@ -57,6 +57,13 @@ "when": 1779582000000, "tag": "0007_tracks_fts", "breakpoints": true + }, + { + "idx": 10, + "version": "7", + "when": 1780000000000, + "tag": "0010_mirror_sources", + "breakpoints": true } ] } \ No newline at end of file diff --git a/apps/api/lexicons/mirror/defs.json b/apps/api/lexicons/mirror/defs.json new file mode 100644 index 00000000..3c951d79 --- /dev/null +++ b/apps/api/lexicons/mirror/defs.json @@ -0,0 +1,42 @@ +{ + "lexicon": 1, + "id": "app.rocksky.mirror.defs", + "defs": { + "mirrorSourceView": { + "type": "object", + "required": [ + "provider", + "enabled", + "hasCredentials" + ], + "properties": { + "provider": { + "type": "string", + "description": "One of: lastfm, listenbrainz, tealfm" + }, + "enabled": { + "type": "boolean", + "description": "Whether scrobbles from this source are being mirrored into Rocksky." + }, + "externalUsername": { + "type": "string", + "description": "Username on the external service (Last.fm / ListenBrainz). Null for Teal.fm." + }, + "hasCredentials": { + "type": "boolean", + "description": "True when an API key is stored. Last.fm/ListenBrainz only; always false for Teal.fm." + }, + "lastPolledAt": { + "type": "string", + "description": "The last time the mirror process successfully polled this source.", + "format": "datetime" + }, + "lastScrobbleSeenAt": { + "type": "string", + "description": "Watermark — scrobbles from the external service older than this are skipped.", + "format": "datetime" + } + } + } + } +} diff --git a/apps/api/lexicons/mirror/getMirrorSources.json b/apps/api/lexicons/mirror/getMirrorSources.json new file mode 100644 index 00000000..ed68bf53 --- /dev/null +++ b/apps/api/lexicons/mirror/getMirrorSources.json @@ -0,0 +1,32 @@ +{ + "lexicon": 1, + "id": "app.rocksky.mirror.getMirrorSources", + "defs": { + "main": { + "type": "query", + "description": "Get the authenticated user's scrobble mirror sources (Last.fm, ListenBrainz, Teal.fm).", + "parameters": { + "type": "params", + "properties": {} + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": [ + "sources" + ], + "properties": { + "sources": { + "type": "array", + "items": { + "type": "ref", + "ref": "app.rocksky.mirror.defs#mirrorSourceView" + } + } + } + } + } + } + } +} diff --git a/apps/api/lexicons/mirror/putMirrorSource.json b/apps/api/lexicons/mirror/putMirrorSource.json new file mode 100644 index 00000000..9c8577ad --- /dev/null +++ b/apps/api/lexicons/mirror/putMirrorSource.json @@ -0,0 +1,44 @@ +{ + "lexicon": 1, + "id": "app.rocksky.mirror.putMirrorSource", + "defs": { + "main": { + "type": "procedure", + "description": "Upsert a mirror source for the authenticated user. Toggling `enabled` notifies the mirror process over NATS so it can start/stop the per-user task without a restart.", + "input": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": [ + "provider" + ], + "properties": { + "provider": { + "type": "string", + "description": "One of: lastfm, listenbrainz, tealfm" + }, + "enabled": { + "type": "boolean", + "description": "Enable or disable mirroring for this provider." + }, + "externalUsername": { + "type": "string", + "description": "External username (Last.fm / ListenBrainz). Required when enabling those providers. Ignored for Teal.fm." + }, + "apiKey": { + "type": "string", + "description": "API key / token to be encrypted at rest. Omit to leave the existing key unchanged. Pass an empty string to clear it." + } + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "ref", + "ref": "app.rocksky.mirror.defs#mirrorSourceView" + } + } + } + } +} diff --git a/apps/api/pkl/defs/mirror/defs.pkl b/apps/api/pkl/defs/mirror/defs.pkl new file mode 100644 index 00000000..54f4dec3 --- /dev/null +++ b/apps/api/pkl/defs/mirror/defs.pkl @@ -0,0 +1,38 @@ +amends "../../schema/lexicon.pkl" + +lexicon = 1 +id = "app.rocksky.mirror.defs" +defs = new Mapping { + ["mirrorSourceView"] { + type = "object" + required = List("provider", "enabled", "hasCredentials") + properties { + ["provider"] = new StringType { + type = "string" + description = "One of: lastfm, listenbrainz, tealfm" + } + ["enabled"] = new BooleanType { + type = "boolean" + description = "Whether scrobbles from this source are being mirrored into Rocksky." + } + ["externalUsername"] = new StringType { + type = "string" + description = "Username on the external service (Last.fm / ListenBrainz). Null for Teal.fm." + } + ["hasCredentials"] = new BooleanType { + type = "boolean" + description = "True when an API key is stored. Last.fm/ListenBrainz only; always false for Teal.fm." + } + ["lastPolledAt"] = new StringType { + type = "string" + format = "datetime" + description = "The last time the mirror process successfully polled this source." + } + ["lastScrobbleSeenAt"] = new StringType { + type = "string" + format = "datetime" + description = "Watermark — scrobbles from the external service older than this are skipped." + } + } + } +} diff --git a/apps/api/pkl/defs/mirror/getMirrorSources.pkl b/apps/api/pkl/defs/mirror/getMirrorSources.pkl new file mode 100644 index 00000000..583b2e3e --- /dev/null +++ b/apps/api/pkl/defs/mirror/getMirrorSources.pkl @@ -0,0 +1,30 @@ +amends "../../schema/lexicon.pkl" + +lexicon = 1 +id = "app.rocksky.mirror.getMirrorSources" +defs = new Mapping { + ["main"] { + type = "query" + description = "Get the authenticated user's scrobble mirror sources (Last.fm, ListenBrainz, Teal.fm)." + parameters { + type = "params" + properties { + } + } + output { + encoding = "application/json" + schema = new ObjectType { + type = "object" + required = List("sources") + properties = new Mapping { + ["sources"] = new Array { + type = "array" + items = new Ref { + ref = "app.rocksky.mirror.defs#mirrorSourceView" + } + } + } + } + } + } +} diff --git a/apps/api/pkl/defs/mirror/putMirrorSource.pkl b/apps/api/pkl/defs/mirror/putMirrorSource.pkl new file mode 100644 index 00000000..4c81d9c1 --- /dev/null +++ b/apps/api/pkl/defs/mirror/putMirrorSource.pkl @@ -0,0 +1,42 @@ +amends "../../schema/lexicon.pkl" + +lexicon = 1 +id = "app.rocksky.mirror.putMirrorSource" +defs = new Mapping { + ["main"] { + type = "procedure" + description = "Upsert a mirror source for the authenticated user. Toggling `enabled` notifies the mirror process over NATS so it can start/stop the per-user task without a restart." + input { + encoding = "application/json" + schema = new ObjectType { + type = "object" + required = List("provider") + properties { + ["provider"] = new StringType { + type = "string" + description = "One of: lastfm, listenbrainz, tealfm" + } + ["enabled"] = new BooleanType { + type = "boolean" + description = "Enable or disable mirroring for this provider." + } + ["externalUsername"] = new StringType { + type = "string" + description = "External username (Last.fm / ListenBrainz). Required when enabling those providers. Ignored for Teal.fm." + } + ["apiKey"] = new StringType { + type = "string" + description = "API key / token to be encrypted at rest. Omit to leave the existing key unchanged. Pass an empty string to clear it." + } + } + } + } + output { + encoding = "application/json" + schema = new Ref { + type = "ref" + ref = "app.rocksky.mirror.defs#mirrorSourceView" + } + } + } +} diff --git a/apps/api/src/lexicon/index.ts b/apps/api/src/lexicon/index.ts index a34dbb12..21d05b57 100644 --- a/apps/api/src/lexicon/index.ts +++ b/apps/api/src/lexicon/index.ts @@ -64,6 +64,8 @@ import type * as AppRockskyLikeDislikeShout from "./types/app/rocksky/like/disli import type * as AppRockskyLikeDislikeSong from "./types/app/rocksky/like/dislikeSong"; import type * as AppRockskyLikeLikeShout from "./types/app/rocksky/like/likeShout"; import type * as AppRockskyLikeLikeSong from "./types/app/rocksky/like/likeSong"; +import type * as AppRockskyMirrorGetMirrorSources from "./types/app/rocksky/mirror/getMirrorSources"; +import type * as AppRockskyMirrorPutMirrorSource from "./types/app/rocksky/mirror/putMirrorSource"; import type * as AppRockskyPlayerAddDirectoryToQueue from "./types/app/rocksky/player/addDirectoryToQueue"; import type * as AppRockskyPlayerAddItemsToQueue from "./types/app/rocksky/player/addItemsToQueue"; import type * as AppRockskyPlayerGetCurrentlyPlaying from "./types/app/rocksky/player/getCurrentlyPlaying"; @@ -253,6 +255,7 @@ export class AppRockskyNS { googledrive: AppRockskyGoogledriveNS; graph: AppRockskyGraphNS; like: AppRockskyLikeNS; + mirror: AppRockskyMirrorNS; player: AppRockskyPlayerNS; playlist: AppRockskyPlaylistNS; scrobble: AppRockskyScrobbleNS; @@ -273,6 +276,7 @@ export class AppRockskyNS { this.googledrive = new AppRockskyGoogledriveNS(server); this.graph = new AppRockskyGraphNS(server); this.like = new AppRockskyLikeNS(server); + this.mirror = new AppRockskyMirrorNS(server); this.player = new AppRockskyPlayerNS(server); this.playlist = new AppRockskyPlaylistNS(server); this.scrobble = new AppRockskyScrobbleNS(server); @@ -913,6 +917,36 @@ export class AppRockskyLikeNS { } } +export class AppRockskyMirrorNS { + _server: Server; + + constructor(server: Server) { + this._server = server; + } + + getMirrorSources( + cfg: ConfigOf< + AV, + AppRockskyMirrorGetMirrorSources.Handler>, + AppRockskyMirrorGetMirrorSources.HandlerReqCtx> + >, + ) { + const nsid = "app.rocksky.mirror.getMirrorSources"; // @ts-ignore + return this._server.xrpc.method(nsid, cfg); + } + + putMirrorSource( + cfg: ConfigOf< + AV, + AppRockskyMirrorPutMirrorSource.Handler>, + AppRockskyMirrorPutMirrorSource.HandlerReqCtx> + >, + ) { + const nsid = "app.rocksky.mirror.putMirrorSource"; // @ts-ignore + return this._server.xrpc.method(nsid, cfg); + } +} + export class AppRockskyPlayerNS { _server: Server; diff --git a/apps/api/src/lexicon/lexicons.ts b/apps/api/src/lexicon/lexicons.ts index 8682ab8a..426ede4f 100644 --- a/apps/api/src/lexicon/lexicons.ts +++ b/apps/api/src/lexicon/lexicons.ts @@ -3976,6 +3976,125 @@ export const schemaDict = { }, }, }, + AppRockskyMirrorDefs: { + lexicon: 1, + id: "app.rocksky.mirror.defs", + defs: { + mirrorSourceView: { + type: "object", + required: ["provider", "enabled", "hasCredentials"], + properties: { + provider: { + type: "string", + description: "One of: lastfm, listenbrainz, tealfm", + }, + enabled: { + type: "boolean", + description: + "Whether scrobbles from this source are being mirrored into Rocksky.", + }, + externalUsername: { + type: "string", + description: + "Username on the external service (Last.fm / ListenBrainz). Null for Teal.fm.", + }, + hasCredentials: { + type: "boolean", + description: + "True when an API key is stored. Last.fm/ListenBrainz only; always false for Teal.fm.", + }, + lastPolledAt: { + type: "string", + description: + "The last time the mirror process successfully polled this source.", + format: "datetime", + }, + lastScrobbleSeenAt: { + type: "string", + description: + "Watermark — scrobbles from the external service older than this are skipped.", + format: "datetime", + }, + }, + }, + }, + }, + AppRockskyMirrorGetMirrorSources: { + lexicon: 1, + id: "app.rocksky.mirror.getMirrorSources", + defs: { + main: { + type: "query", + description: + "Get the authenticated user's scrobble mirror sources (Last.fm, ListenBrainz, Teal.fm).", + parameters: { + type: "params", + properties: {}, + }, + output: { + encoding: "application/json", + schema: { + type: "object", + required: ["sources"], + properties: { + sources: { + type: "array", + items: { + type: "ref", + ref: "lex:app.rocksky.mirror.defs#mirrorSourceView", + }, + }, + }, + }, + }, + }, + }, + }, + AppRockskyMirrorPutMirrorSource: { + lexicon: 1, + id: "app.rocksky.mirror.putMirrorSource", + defs: { + main: { + type: "procedure", + description: + "Upsert a mirror source for the authenticated user. Toggling `enabled` notifies the mirror process over NATS so it can start/stop the per-user task without a restart.", + input: { + encoding: "application/json", + schema: { + type: "object", + required: ["provider"], + properties: { + provider: { + type: "string", + description: "One of: lastfm, listenbrainz, tealfm", + }, + enabled: { + type: "boolean", + description: "Enable or disable mirroring for this provider.", + }, + externalUsername: { + type: "string", + description: + "External username (Last.fm / ListenBrainz). Required when enabling those providers. Ignored for Teal.fm.", + }, + apiKey: { + type: "string", + description: + "API key / token to be encrypted at rest. Omit to leave the existing key unchanged. Pass an empty string to clear it.", + }, + }, + }, + }, + output: { + encoding: "application/json", + schema: { + type: "ref", + ref: "lex:app.rocksky.mirror.defs#mirrorSourceView", + }, + }, + }, + }, + }, AppRockskyPlayerAddDirectoryToQueue: { lexicon: 1, id: "app.rocksky.player.addDirectoryToQueue", @@ -6957,6 +7076,9 @@ export const ids = { AppRockskyLike: "app.rocksky.like", AppRockskyLikeLikeShout: "app.rocksky.like.likeShout", AppRockskyLikeLikeSong: "app.rocksky.like.likeSong", + AppRockskyMirrorDefs: "app.rocksky.mirror.defs", + AppRockskyMirrorGetMirrorSources: "app.rocksky.mirror.getMirrorSources", + AppRockskyMirrorPutMirrorSource: "app.rocksky.mirror.putMirrorSource", AppRockskyPlayerAddDirectoryToQueue: "app.rocksky.player.addDirectoryToQueue", AppRockskyPlayerAddItemsToQueue: "app.rocksky.player.addItemsToQueue", AppRockskyPlayerDefs: "app.rocksky.player.defs", diff --git a/apps/api/src/lexicon/types/app/rocksky/mirror/defs.ts b/apps/api/src/lexicon/types/app/rocksky/mirror/defs.ts new file mode 100644 index 00000000..9fe0303b --- /dev/null +++ b/apps/api/src/lexicon/types/app/rocksky/mirror/defs.ts @@ -0,0 +1,35 @@ +/** + * GENERATED CODE - DO NOT MODIFY + */ +import { type ValidationResult, BlobRef } from "@atproto/lexicon"; +import { lexicons } from "../../../../lexicons"; +import { isObj, hasProp } from "../../../../util"; +import { CID } from "multiformats/cid"; + +export interface MirrorSourceView { + /** One of: lastfm, listenbrainz, tealfm */ + provider: string; + /** Whether scrobbles from this source are being mirrored into Rocksky. */ + enabled: boolean; + /** Username on the external service (Last.fm / ListenBrainz). Null for Teal.fm. */ + externalUsername?: string; + /** True when an API key is stored. Last.fm/ListenBrainz only; always false for Teal.fm. */ + hasCredentials: boolean; + /** The last time the mirror process successfully polled this source. */ + lastPolledAt?: string; + /** Watermark — scrobbles from the external service older than this are skipped. */ + lastScrobbleSeenAt?: string; + [k: string]: unknown; +} + +export function isMirrorSourceView(v: unknown): v is MirrorSourceView { + return ( + isObj(v) && + hasProp(v, "$type") && + v.$type === "app.rocksky.mirror.defs#mirrorSourceView" + ); +} + +export function validateMirrorSourceView(v: unknown): ValidationResult { + return lexicons.validate("app.rocksky.mirror.defs#mirrorSourceView", v); +} diff --git a/apps/api/src/lexicon/types/app/rocksky/mirror/getMirrorSources.ts b/apps/api/src/lexicon/types/app/rocksky/mirror/getMirrorSources.ts new file mode 100644 index 00000000..287e5214 --- /dev/null +++ b/apps/api/src/lexicon/types/app/rocksky/mirror/getMirrorSources.ts @@ -0,0 +1,45 @@ +/** + * GENERATED CODE - DO NOT MODIFY + */ +import type express from "express"; +import { ValidationResult, BlobRef } from "@atproto/lexicon"; +import { lexicons } from "../../../../lexicons"; +import { isObj, hasProp } from "../../../../util"; +import { CID } from "multiformats/cid"; +import type { HandlerAuth, HandlerPipeThrough } from "@atproto/xrpc-server"; +import type * as AppRockskyMirrorDefs from "./defs"; + +export type QueryParams = {}; + +export type InputSchema = undefined; + +export interface OutputSchema { + sources: AppRockskyMirrorDefs.MirrorSourceView[]; + [k: string]: unknown; +} + +export type HandlerInput = undefined; + +export interface HandlerSuccess { + encoding: "application/json"; + body: OutputSchema; + headers?: { [key: string]: string }; +} + +export interface HandlerError { + status: number; + message?: string; +} + +export type HandlerOutput = HandlerError | HandlerSuccess | HandlerPipeThrough; +export type HandlerReqCtx = { + auth: HA; + params: QueryParams; + input: HandlerInput; + req: express.Request; + res: express.Response; + resetRouteRateLimits: () => Promise; +}; +export type Handler = ( + ctx: HandlerReqCtx, +) => Promise | HandlerOutput; diff --git a/apps/api/src/lexicon/types/app/rocksky/mirror/putMirrorSource.ts b/apps/api/src/lexicon/types/app/rocksky/mirror/putMirrorSource.ts new file mode 100644 index 00000000..7aae4561 --- /dev/null +++ b/apps/api/src/lexicon/types/app/rocksky/mirror/putMirrorSource.ts @@ -0,0 +1,55 @@ +/** + * GENERATED CODE - DO NOT MODIFY + */ +import type express from "express"; +import { ValidationResult, BlobRef } from "@atproto/lexicon"; +import { lexicons } from "../../../../lexicons"; +import { isObj, hasProp } from "../../../../util"; +import { CID } from "multiformats/cid"; +import type { HandlerAuth, HandlerPipeThrough } from "@atproto/xrpc-server"; +import type * as AppRockskyMirrorDefs from "./defs"; + +export type QueryParams = {}; + +export interface InputSchema { + /** One of: lastfm, listenbrainz, tealfm */ + provider: string; + /** Enable or disable mirroring for this provider. */ + enabled?: boolean; + /** External username (Last.fm / ListenBrainz). Required when enabling those providers. Ignored for Teal.fm. */ + externalUsername?: string; + /** API key / token to be encrypted at rest. Omit to leave the existing key unchanged. Pass an empty string to clear it. */ + apiKey?: string; + [k: string]: unknown; +} + +export type OutputSchema = AppRockskyMirrorDefs.MirrorSourceView; + +export interface HandlerInput { + encoding: "application/json"; + body: InputSchema; +} + +export interface HandlerSuccess { + encoding: "application/json"; + body: OutputSchema; + headers?: { [key: string]: string }; +} + +export interface HandlerError { + status: number; + message?: string; +} + +export type HandlerOutput = HandlerError | HandlerSuccess | HandlerPipeThrough; +export type HandlerReqCtx = { + auth: HA; + params: QueryParams; + input: HandlerInput; + req: express.Request; + res: express.Response; + resetRouteRateLimits: () => Promise; +}; +export type Handler = ( + ctx: HandlerReqCtx, +) => Promise | HandlerOutput; diff --git a/apps/api/src/schema/index.ts b/apps/api/src/schema/index.ts index 813274f0..d64eb878 100644 --- a/apps/api/src/schema/index.ts +++ b/apps/api/src/schema/index.ts @@ -17,6 +17,7 @@ import googleDriveDirectories from "./google-drive-directories"; import googleDrivePaths from "./google-drive-paths"; import googleDrive from "./googledrive"; import lovedTracks from "./loved-tracks"; +import mirrorSources from "./mirror-sources"; import playlistTracks from "./playlist-tracks"; import playlists from "./playlists"; import profileShouts from "./profile-shouts"; @@ -80,4 +81,5 @@ export default { queueTracks, feeds, follows, + mirrorSources, }; diff --git a/apps/api/src/schema/mirror-sources.ts b/apps/api/src/schema/mirror-sources.ts new file mode 100644 index 00000000..a6c9f78f --- /dev/null +++ b/apps/api/src/schema/mirror-sources.ts @@ -0,0 +1,39 @@ +import { type InferInsertModel, type InferSelectModel, sql } from "drizzle-orm"; +import { + boolean, + index, + integer, + pgTable, + text, + timestamp, + uniqueIndex, +} from "drizzle-orm/pg-core"; +import users from "./users"; + +const mirrorSources = pgTable( + "mirror_sources", + { + id: text("xata_id").primaryKey().default(sql`xata_id()`), + userId: text("user_id") + .notNull() + .references(() => users.id), + provider: text("provider").notNull(), + enabled: boolean("enabled").notNull().default(false), + externalUsername: text("external_username"), + encryptedApiKey: text("encrypted_api_key"), + lastPolledAt: timestamp("last_polled_at"), + lastScrobbleSeenAt: timestamp("last_scrobble_seen_at"), + createdAt: timestamp("xata_createdat").defaultNow().notNull(), + updatedAt: timestamp("xata_updatedat").defaultNow().notNull(), + xataVersion: integer("xata_version"), + }, + (t) => [ + uniqueIndex("mirror_sources_user_provider_idx").on(t.userId, t.provider), + index("mirror_sources_enabled_provider_idx").on(t.enabled, t.provider), + ], +); + +export type SelectMirrorSource = InferSelectModel; +export type InsertMirrorSource = InferInsertModel; + +export default mirrorSources; diff --git a/apps/api/src/xrpc/app/rocksky/mirror/getMirrorSources.ts b/apps/api/src/xrpc/app/rocksky/mirror/getMirrorSources.ts new file mode 100644 index 00000000..f0a31318 --- /dev/null +++ b/apps/api/src/xrpc/app/rocksky/mirror/getMirrorSources.ts @@ -0,0 +1,81 @@ +import type { HandlerAuth } from "@atproto/xrpc-server"; +import { consola } from "consola"; +import type { Context } from "context"; +import { eq } from "drizzle-orm"; +import { Effect, pipe } from "effect"; +import type { Server } from "lexicon"; +import type { MirrorSourceView } from "lexicon/types/app/rocksky/mirror/defs"; +import tables from "schema"; + +const PROVIDERS = ["lastfm", "listenbrainz", "tealfm"] as const; + +export default function (server: Server, ctx: Context) { + const getMirrorSources = (auth: HandlerAuth) => + pipe( + { ctx, did: auth.credentials?.did as string | undefined }, + retrieve, + Effect.flatMap(presentation), + Effect.retry({ times: 3 }), + Effect.timeout("10 seconds"), + Effect.catchAll((err) => { + consola.error(err); + return Effect.succeed({ sources: emptySources() }); + }), + ); + + server.app.rocksky.mirror.getMirrorSources({ + auth: ctx.authVerifier, + handler: async ({ auth }) => { + const result = await Effect.runPromise(getMirrorSources(auth)); + return { encoding: "application/json" as const, body: result }; + }, + }); +} + +const retrieve = ({ + ctx, + did, +}: { + ctx: Context; + did: string | undefined; +}): Effect.Effect => + Effect.tryPromise({ + try: async () => { + if (!did) return []; + const [user] = await ctx.db + .select({ id: tables.users.id }) + .from(tables.users) + .where(eq(tables.users.did, did)) + .limit(1); + if (!user) return []; + + const rows = await ctx.db + .select() + .from(tables.mirrorSources) + .where(eq(tables.mirrorSources.userId, user.id)); + + const byProvider = new Map(rows.map((r) => [r.provider, r])); + return PROVIDERS.map((provider) => { + const r = byProvider.get(provider); + return { + provider, + enabled: r?.enabled ?? false, + externalUsername: r?.externalUsername ?? undefined, + hasCredentials: !!r?.encryptedApiKey, + lastPolledAt: r?.lastPolledAt?.toISOString(), + lastScrobbleSeenAt: r?.lastScrobbleSeenAt?.toISOString(), + } satisfies MirrorSourceView; + }); + }, + catch: (error) => new Error(`Failed to retrieve mirror sources: ${error}`), + }); + +const presentation = (sources: MirrorSourceView[]) => + Effect.sync(() => ({ sources })); + +const emptySources = (): MirrorSourceView[] => + PROVIDERS.map((provider) => ({ + provider, + enabled: false, + hasCredentials: false, + })); diff --git a/apps/api/src/xrpc/app/rocksky/mirror/putMirrorSource.ts b/apps/api/src/xrpc/app/rocksky/mirror/putMirrorSource.ts new file mode 100644 index 00000000..879d9315 --- /dev/null +++ b/apps/api/src/xrpc/app/rocksky/mirror/putMirrorSource.ts @@ -0,0 +1,149 @@ +import type { HandlerAuth } from "@atproto/xrpc-server"; +import { consola } from "consola"; +import type { Context } from "context"; +import { and, eq } from "drizzle-orm"; +import { Effect, pipe } from "effect"; +import type { Server } from "lexicon"; +import type { MirrorSourceView } from "lexicon/types/app/rocksky/mirror/defs"; +import type { InputSchema } from "lexicon/types/app/rocksky/mirror/putMirrorSource"; +import { encryptCredential } from "lib/storage-crypto"; +import tables from "schema"; + +const PROVIDERS = new Set(["lastfm", "listenbrainz", "tealfm"]); +const MIRROR_TOPIC = "rocksky.mirror.user"; + +export default function (server: Server, ctx: Context) { + const putMirrorSource = (input: InputSchema, auth: HandlerAuth) => + pipe( + { ctx, input, did: auth.credentials?.did as string | undefined }, + upsert, + Effect.flatMap(presentation), + Effect.timeout("10 seconds"), + Effect.catchAll((err) => { + consola.error(err); + return Effect.succeed({ + provider: input.provider, + enabled: false, + hasCredentials: false, + } satisfies MirrorSourceView); + }), + ); + + server.app.rocksky.mirror.putMirrorSource({ + auth: ctx.authVerifier, + handler: async ({ input, auth }) => { + const result = await Effect.runPromise(putMirrorSource(input.body, auth)); + return { encoding: "application/json" as const, body: result }; + }, + }); +} + +const upsert = ({ + ctx, + input, + did, +}: { + ctx: Context; + input: InputSchema; + did: string | undefined; +}): Effect.Effect => + Effect.tryPromise({ + try: async () => { + if (!did) throw new Error("Unauthorized"); + if (!PROVIDERS.has(input.provider)) { + throw new Error(`Unknown provider: ${input.provider}`); + } + + const [user] = await ctx.db + .select({ id: tables.users.id }) + .from(tables.users) + .where(eq(tables.users.did, did)) + .limit(1); + if (!user) throw new Error("User not found"); + + const [existing] = await ctx.db + .select() + .from(tables.mirrorSources) + .where( + and( + eq(tables.mirrorSources.userId, user.id), + eq(tables.mirrorSources.provider, input.provider), + ), + ) + .limit(1); + + // Last.fm / ListenBrainz require an external username before enabling. + if ( + input.enabled === true && + (input.provider === "lastfm" || input.provider === "listenbrainz") && + !(input.externalUsername ?? existing?.externalUsername) + ) { + throw new Error("externalUsername is required"); + } + + const encryptedApiKey = + input.apiKey === undefined + ? undefined + : input.apiKey === "" + ? null + : await encryptCredential(input.apiKey); + + // Bump watermark on (re-)enable so we only mirror new plays. + const enableTransition = + input.enabled === true && existing?.enabled !== true; + const now = new Date(); + + let row: typeof tables.mirrorSources.$inferSelect; + if (existing) { + const updates: Record = { updatedAt: now }; + if (input.enabled !== undefined) updates.enabled = input.enabled; + if (input.externalUsername !== undefined) + updates.externalUsername = input.externalUsername; + if (encryptedApiKey !== undefined) + updates.encryptedApiKey = encryptedApiKey; + if (enableTransition) updates.lastScrobbleSeenAt = now; + + const [updated] = await ctx.db + .update(tables.mirrorSources) + .set(updates) + .where(eq(tables.mirrorSources.id, existing.id)) + .returning(); + row = updated; + } else { + const [inserted] = await ctx.db + .insert(tables.mirrorSources) + .values({ + userId: user.id, + provider: input.provider, + enabled: input.enabled ?? false, + externalUsername: input.externalUsername ?? null, + encryptedApiKey: encryptedApiKey ?? null, + lastScrobbleSeenAt: input.enabled ? now : null, + }) + .returning(); + row = inserted; + } + + // Best-effort notify the mirror process to start/stop the per-user task. + try { + ctx.nc.publish( + MIRROR_TOPIC, + Buffer.from(`${input.provider}:${user.id}`), + ); + } catch (e) { + consola.warn("[putMirrorSource] NATS publish failed:", e); + } + + return { + provider: row.provider, + enabled: row.enabled, + externalUsername: row.externalUsername ?? undefined, + hasCredentials: !!row.encryptedApiKey, + lastPolledAt: row.lastPolledAt?.toISOString(), + lastScrobbleSeenAt: row.lastScrobbleSeenAt?.toISOString(), + } satisfies MirrorSourceView; + }, + catch: (error) => new Error(`Failed to upsert mirror source: ${error}`), + }); + +const presentation = (view: MirrorSourceView) => Effect.sync(() => view); diff --git a/apps/api/src/xrpc/index.ts b/apps/api/src/xrpc/index.ts index 0c5b2534..4bd853d3 100644 --- a/apps/api/src/xrpc/index.ts +++ b/apps/api/src/xrpc/index.ts @@ -28,6 +28,8 @@ import search from "./app/rocksky/feed/search"; import downloadFileFromGoogleDrive from "./app/rocksky/googledrive/downloadFile"; import getFileFromGoogleDrive from "./app/rocksky/googledrive/getFile"; import getFilesFromGoogleDrive from "./app/rocksky/googledrive/getFiles"; +import getMirrorSources from "./app/rocksky/mirror/getMirrorSources"; +import putMirrorSource from "./app/rocksky/mirror/putMirrorSource"; import dislikeShout from "./app/rocksky/like/dislikeShout"; import dislikeSong from "./app/rocksky/like/dislikeSong"; import likeShout from "./app/rocksky/like/likeShout"; @@ -181,6 +183,8 @@ export default function (server: Server, ctx: Context) { getRecommendations(server, ctx); getArtistRecommendations(server, ctx); getAlbumRecommendations(server, ctx); + getMirrorSources(server, ctx); + putMirrorSource(server, ctx); return server; } diff --git a/apps/web-mobile/src/App.tsx b/apps/web-mobile/src/App.tsx index eb1b4990..3806903c 100644 --- a/apps/web-mobile/src/App.tsx +++ b/apps/web-mobile/src/App.tsx @@ -2,6 +2,7 @@ import { BrowserRouter, Route, Routes } from "react-router-dom"; import MiniPlayer from "./components/MiniPlayer"; import AlbumPage from "./pages/album"; import ApiKeysPage from "./pages/apikeys"; +import MirrorsPage from "./pages/mirrors"; import StoragePage from "./pages/storage"; import ArtistPage from "./pages/artist"; import Charts from "./pages/charts"; @@ -30,6 +31,7 @@ function App() { } /> } /> } /> + } /> } /> } /> } /> diff --git a/apps/web-mobile/src/api/mirror.ts b/apps/web-mobile/src/api/mirror.ts new file mode 100644 index 00000000..976325ca --- /dev/null +++ b/apps/web-mobile/src/api/mirror.ts @@ -0,0 +1,42 @@ +import { client } from "."; + +export type MirrorProvider = "lastfm" | "listenbrainz" | "tealfm"; + +export interface MirrorSourceView { + provider: MirrorProvider; + enabled: boolean; + externalUsername?: string; + hasCredentials: boolean; + lastPolledAt?: string; + lastScrobbleSeenAt?: string; +} + +const authHeader = () => ({ + Authorization: `Bearer ${localStorage.getItem("token")}`, +}); + +export const getMirrorSources = async (): Promise => { + const res = await client.get<{ sources: MirrorSourceView[] }>( + "/xrpc/app.rocksky.mirror.getMirrorSources", + { headers: authHeader() }, + ); + return res.data.sources ?? []; +}; + +export interface PutMirrorSourceInput { + provider: MirrorProvider; + enabled?: boolean; + externalUsername?: string; + apiKey?: string; +} + +export const putMirrorSource = async ( + input: PutMirrorSourceInput, +): Promise => { + const res = await client.post( + "/xrpc/app.rocksky.mirror.putMirrorSource", + input, + { headers: authHeader() }, + ); + return res.data; +}; diff --git a/apps/web-mobile/src/layouts/Navbar/Navbar.tsx b/apps/web-mobile/src/layouts/Navbar/Navbar.tsx index 861be147..de813059 100644 --- a/apps/web-mobile/src/layouts/Navbar/Navbar.tsx +++ b/apps/web-mobile/src/layouts/Navbar/Navbar.tsx @@ -125,14 +125,24 @@ function Navbar() { API Keys {profile?.did === "did:plc:7vdlgi2bflelz7mmuxoqjfcr" && ( - setMenuOpen(false)} - > - Storage - + <> + setMenuOpen(false)} + > + Storage + + setMenuOpen(false)} + > + Mirror sources + + )} + + + )} + + {source.provider === "tealfm" && ( +

+ No credentials required — your Bluesky DID is enough. +

+ )} + + ); +} + +export default function MirrorsPage() { + const navigate = useNavigate(); + const queryClient = useQueryClient(); + const jwt = localStorage.getItem("token"); + + const { data: sources = [] } = useQuery({ + queryKey: ["mirror-sources"], + queryFn: getMirrorSources, + enabled: !!jwt, + }); + + const updateMutation = useMutation({ + mutationFn: putMirrorSource, + onSuccess: () => + queryClient.invalidateQueries({ queryKey: ["mirror-sources"] }), + }); + + if (!jwt) { + navigate("/"); + return null; + } + + if (localStorage.getItem("did") !== "did:plc:7vdlgi2bflelz7mmuxoqjfcr") { + navigate("/"); + return null; + } + + return ( +
+
+

+ Mirror sources +

+ +
+ +

+ Tokens are encrypted at rest using XSalsa20-Poly1305 and never + returned by the API. Mirrored scrobbles are deduplicated against + existing ones within a ±120 second window. +

+
+ + {sources.map((s) => ( + + updateMutation.mutate({ provider: s.provider, ...input }) + } + /> + ))} +
+
+ ); +} diff --git a/apps/web/src/api/mirror.ts b/apps/web/src/api/mirror.ts new file mode 100644 index 00000000..2696603e --- /dev/null +++ b/apps/web/src/api/mirror.ts @@ -0,0 +1,43 @@ +import { client } from "."; + +export type MirrorProvider = "lastfm" | "listenbrainz" | "tealfm"; + +export interface MirrorSourceView { + provider: MirrorProvider; + enabled: boolean; + externalUsername?: string; + hasCredentials: boolean; + lastPolledAt?: string; + lastScrobbleSeenAt?: string; +} + +const authHeader = () => ({ + Authorization: `Bearer ${localStorage.getItem("token")}`, +}); + +export const getMirrorSources = async (): Promise => { + const res = await client.get<{ sources: MirrorSourceView[] }>( + "/xrpc/app.rocksky.mirror.getMirrorSources", + { headers: authHeader() }, + ); + return res.data.sources ?? []; +}; + +export interface PutMirrorSourceInput { + provider: MirrorProvider; + enabled?: boolean; + externalUsername?: string; + /** Omit to leave existing key unchanged. Empty string to clear. */ + apiKey?: string; +} + +export const putMirrorSource = async ( + input: PutMirrorSourceInput, +): Promise => { + const res = await client.post( + "/xrpc/app.rocksky.mirror.putMirrorSource", + input, + { headers: authHeader() }, + ); + return res.data; +}; diff --git a/apps/web/src/layouts/Navbar/Navbar.tsx b/apps/web/src/layouts/Navbar/Navbar.tsx index cde59a86..20fb9a6f 100644 --- a/apps/web/src/layouts/Navbar/Navbar.tsx +++ b/apps/web/src/layouts/Navbar/Navbar.tsx @@ -306,6 +306,14 @@ function Navbar() { ), }, + { + id: "mirrors", + label: ( + + Mirror sources + + ), + }, ] : []), ...(profile?.did === "did:plc:7vdlgi2bflelz7mmuxoqjfcr" @@ -387,6 +395,9 @@ function Navbar() { case "storage": navigate({ to: "/storage" }); break; + case "mirrors": + navigate({ to: "/mirrors" }); + break; case "import-history": navigate({ to: "/import", diff --git a/apps/web/src/pages/mirrors/Mirrors.tsx b/apps/web/src/pages/mirrors/Mirrors.tsx new file mode 100644 index 00000000..5aa55606 --- /dev/null +++ b/apps/web/src/pages/mirrors/Mirrors.tsx @@ -0,0 +1,251 @@ +import { IconBroadcast, IconLock } from "@tabler/icons-react"; +import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; +import { useNavigate } from "@tanstack/react-router"; +import { Button } from "baseui/button"; +import { Checkbox, STYLE_TYPE } from "baseui/checkbox"; +import { FormControl } from "baseui/form-control"; +import { Input } from "baseui/input"; +import { LabelSmall } from "baseui/typography"; +import { useAtomValue } from "jotai"; +import { useEffect, useState } from "react"; +import { + MirrorProvider, + MirrorSourceView, + getMirrorSources, + putMirrorSource, +} from "../../api/mirror"; +import { profileAtom } from "../../atoms/profile"; +import Main from "../../layouts/Main"; + +const inputOverrides = { + Root: { + style: { + backgroundColor: "var(--color-input-background)", + borderColor: "var(--color-input-background)", + }, + }, + InputContainer: { + style: { backgroundColor: "var(--color-input-background)" }, + }, + Input: { + style: { color: "var(--color-text)", caretColor: "var(--color-text)" }, + }, +}; + +const PROVIDER_LABEL: Record = { + lastfm: "Last.fm", + listenbrainz: "ListenBrainz", + tealfm: "Teal.fm", +}; + +const PROVIDER_DESCRIPTION: Record = { + lastfm: + "Poll your Last.fm recent tracks every 30s and mirror new scrobbles into Rocksky.", + listenbrainz: + "Poll your ListenBrainz listens every 30s and mirror new ones into Rocksky.", + tealfm: + "Listen to your Teal.fm play events on Jetstream and mirror them into Rocksky in real time.", +}; + +function SourceCard({ + source, + onChange, + busy, +}: { + source: MirrorSourceView; + onChange: (input: { + enabled?: boolean; + externalUsername?: string; + apiKey?: string; + }) => void; + busy: boolean; +}) { + const needsCredentials = + source.provider === "lastfm" || source.provider === "listenbrainz"; + const [username, setUsername] = useState(source.externalUsername ?? ""); + const [apiKey, setApiKey] = useState(""); + + // Keep local state in sync if the row is refetched. + useEffect(() => { + setUsername(source.externalUsername ?? ""); + }, [source.externalUsername]); + + return ( +
+
+
+ +
+

+ {PROVIDER_LABEL[source.provider]} +

+

+ {PROVIDER_DESCRIPTION[source.provider]} +

+
+
+ onChange({ enabled: e.target.checked })} + /> +
+ + {needsCredentials && ( +
+ + setUsername(e.currentTarget.value)} + placeholder={ + source.provider === "lastfm" ? "lastfm-username" : "listenbrainz-username" + } + overrides={inputOverrides} + /> + + + setApiKey(e.currentTarget.value)} + type="password" + placeholder={source.hasCredentials ? "••••••••" : "Paste your token"} + overrides={inputOverrides} + /> + +
+ +
+
+ )} + + {source.provider === "tealfm" && ( + + No credentials required — your Bluesky DID is enough. + + )} +
+ ); +} + +export default function MirrorsPage() { + const profile = useAtomValue(profileAtom); + const navigate = useNavigate(); + const queryClient = useQueryClient(); + const jwt = localStorage.getItem("token"); + + const { data: sources = [] } = useQuery({ + queryKey: ["mirror-sources"], + queryFn: getMirrorSources, + enabled: !!jwt, + }); + + const updateMutation = useMutation({ + mutationFn: putMirrorSource, + onSuccess: () => + queryClient.invalidateQueries({ queryKey: ["mirror-sources"] }), + }); + + if (!jwt || !profile) { + navigate({ to: "/" }); + return null; + } + + if (profile.did !== "did:plc:7vdlgi2bflelz7mmuxoqjfcr") { + navigate({ to: "/" }); + return null; + } + + return ( +
+
+

+ Mirror sources +

+ +
+ + + Tokens are encrypted at rest using XSalsa20-Poly1305 and never + returned by the API. Mirrored scrobbles are deduplicated against + existing ones within a ±120 second window. + +
+ + {sources.map((s) => ( + + updateMutation.mutate({ provider: s.provider, ...input }) + } + /> + ))} +
+
+ ); +} diff --git a/apps/web/src/pages/mirrors/index.tsx b/apps/web/src/pages/mirrors/index.tsx new file mode 100644 index 00000000..0d3b3f67 --- /dev/null +++ b/apps/web/src/pages/mirrors/index.tsx @@ -0,0 +1,3 @@ +import MirrorsPage from "./Mirrors"; + +export default MirrorsPage; diff --git a/apps/web/src/routeTree.gen.ts b/apps/web/src/routeTree.gen.ts index e20bba28..b8412920 100644 --- a/apps/web/src/routeTree.gen.ts +++ b/apps/web/src/routeTree.gen.ts @@ -14,6 +14,7 @@ import { Route as TosRouteImport } from './routes/tos' import { Route as StorageRouteImport } from './routes/storage' import { Route as ScrobbleRouteImport } from './routes/scrobble' import { Route as RecommendationsRouteImport } from './routes/recommendations' +import { Route as MirrorsRouteImport } from './routes/mirrors' import { Route as LoadingRouteImport } from './routes/loading' import { Route as ImportRouteImport } from './routes/import' import { Route as ChartsRouteImport } from './routes/charts' @@ -68,6 +69,11 @@ const RecommendationsRoute = RecommendationsRouteImport.update({ path: '/recommendations', getParentRoute: () => rootRouteImport, } as any) +const MirrorsRoute = MirrorsRouteImport.update({ + id: '/mirrors', + path: '/mirrors', + getParentRoute: () => rootRouteImport, +} as any) const LoadingRoute = LoadingRouteImport.update({ id: '/loading', path: '/loading', @@ -215,6 +221,7 @@ export interface FileRoutesByFullPath { '/charts': typeof ChartsRoute '/import': typeof ImportRoute '/loading': typeof LoadingRoute + '/mirrors': typeof MirrorsRoute '/recommendations': typeof RecommendationsRoute '/scrobble': typeof ScrobbleRoute '/storage': typeof StorageRoute @@ -250,6 +257,7 @@ export interface FileRoutesByTo { '/charts': typeof ChartsRoute '/import': typeof ImportRoute '/loading': typeof LoadingRoute + '/mirrors': typeof MirrorsRoute '/recommendations': typeof RecommendationsRoute '/scrobble': typeof ScrobbleRoute '/storage': typeof StorageRoute @@ -286,6 +294,7 @@ export interface FileRoutesById { '/charts': typeof ChartsRoute '/import': typeof ImportRoute '/loading': typeof LoadingRoute + '/mirrors': typeof MirrorsRoute '/recommendations': typeof RecommendationsRoute '/scrobble': typeof ScrobbleRoute '/storage': typeof StorageRoute @@ -323,6 +332,7 @@ export interface FileRouteTypes { | '/charts' | '/import' | '/loading' + | '/mirrors' | '/recommendations' | '/scrobble' | '/storage' @@ -358,6 +368,7 @@ export interface FileRouteTypes { | '/charts' | '/import' | '/loading' + | '/mirrors' | '/recommendations' | '/scrobble' | '/storage' @@ -393,6 +404,7 @@ export interface FileRouteTypes { | '/charts' | '/import' | '/loading' + | '/mirrors' | '/recommendations' | '/scrobble' | '/storage' @@ -429,6 +441,7 @@ export interface RootRouteChildren { ChartsRoute: typeof ChartsRoute ImportRoute: typeof ImportRoute LoadingRoute: typeof LoadingRoute + MirrorsRoute: typeof MirrorsRoute RecommendationsRoute: typeof RecommendationsRoute ScrobbleRoute: typeof ScrobbleRoute StorageRoute: typeof StorageRoute @@ -496,6 +509,13 @@ declare module '@tanstack/react-router' { preLoaderRoute: typeof RecommendationsRouteImport parentRoute: typeof rootRouteImport } + '/mirrors': { + id: '/mirrors' + path: '/mirrors' + fullPath: '/mirrors' + preLoaderRoute: typeof MirrorsRouteImport + parentRoute: typeof rootRouteImport + } '/loading': { id: '/loading' path: '/loading' @@ -701,6 +721,7 @@ const rootRouteChildren: RootRouteChildren = { ChartsRoute: ChartsRoute, ImportRoute: ImportRoute, LoadingRoute: LoadingRoute, + MirrorsRoute: MirrorsRoute, RecommendationsRoute: RecommendationsRoute, ScrobbleRoute: ScrobbleRoute, StorageRoute: StorageRoute, diff --git a/apps/web/src/routes/mirrors.tsx b/apps/web/src/routes/mirrors.tsx new file mode 100644 index 00000000..a5659c21 --- /dev/null +++ b/apps/web/src/routes/mirrors.tsx @@ -0,0 +1,6 @@ +import { createFileRoute } from "@tanstack/react-router"; +import MirrorsPage from "../pages/mirrors"; + +export const Route = createFileRoute("/mirrors")({ + component: MirrorsPage, +}); diff --git a/crates/mirror/Cargo.toml b/crates/mirror/Cargo.toml new file mode 100644 index 00000000..80497726 --- /dev/null +++ b/crates/mirror/Cargo.toml @@ -0,0 +1,42 @@ +[package] +name = "rocksky-mirror" +version = "0.1.0" +authors.workspace = true +edition.workspace = true +license.workspace = true +repository.workspace = true + +[dependencies] +anyhow = "1.0.98" +async-nats = "0.39.0" +base64 = "0.22.1" +chrono = { version = "= 0.4.39", features = ["serde"] } +dotenv = "0.15.0" +dryoc = "0.6.2" +futures-util = "0.3.31" +hex = "0.4.3" +jsonwebtoken = "9.3.1" +owo-colors = "4.1.0" +reqwest = { version = "0.12.12", features = [ + "rustls-tls", + "json", +], default-features = false } +serde = { version = "1.0.219", features = ["derive"] } +serde_json = "1.0.140" +sqlx = { version = "0.8.3", features = [ + "runtime-tokio", + "tls-rustls", + "postgres", + "chrono", + "derive", + "macros", +] } +tokio = { version = "1.43.0", features = ["full"] } +tokio-stream = { version = "0.1.17", features = ["full"] } +tokio-tungstenite = { version = "0.26.2", features = [ + "tokio-rustls", + "rustls-tls-webpki-roots", +] } +tokio-util = { version = "0.7.13", features = ["rt"] } +tracing = "0.1.41" +tracing-subscriber = { version = "0.3.19", features = ["env-filter"] } diff --git a/crates/mirror/src/crypto.rs b/crates/mirror/src/crypto.rs new file mode 100644 index 00000000..e32506a8 --- /dev/null +++ b/crates/mirror/src/crypto.rs @@ -0,0 +1,52 @@ +//! Decrypts API keys stored by `apps/api/src/lib/storage-crypto.ts`. +//! +//! Layout: base64( nonce (24 bytes) || ciphertext_with_tag ) +//! Algorithm: libsodium `crypto_secretbox_easy` (XSalsa20-Poly1305). +//! libsodium-wrappers' `to_base64()` defaults to URL-safe + no padding. + +use anyhow::{anyhow, Context, Error}; +use base64::{ + engine::general_purpose::{STANDARD, STANDARD_NO_PAD, URL_SAFE, URL_SAFE_NO_PAD}, + Engine, +}; +use dryoc::classic::crypto_secretbox::{crypto_secretbox_open_easy, Key, Nonce}; +use std::env; + +const NONCE_BYTES: usize = 24; + +pub fn decrypt(encoded: &str) -> Result { + let key_hex = env::var("STORAGE_ENCRYPTION_KEY").context("STORAGE_ENCRYPTION_KEY not set")?; + let key_bytes = hex::decode(&key_hex).context("STORAGE_ENCRYPTION_KEY must be hex")?; + if key_bytes.len() != 32 { + return Err(anyhow!( + "STORAGE_ENCRYPTION_KEY must decode to 32 bytes, got {}", + key_bytes.len() + )); + } + let mut key: Key = [0u8; 32]; + key.copy_from_slice(&key_bytes); + + let combined = decode_base64_permissive(encoded)?; + const TAG_BYTES: usize = 16; + if combined.len() <= NONCE_BYTES + TAG_BYTES { + return Err(anyhow!("ciphertext too short ({} bytes)", combined.len())); + } + let (nonce_bytes, ct) = combined.split_at(NONCE_BYTES); + + let mut nonce: Nonce = [0u8; 24]; + nonce.copy_from_slice(nonce_bytes); + + let mut plaintext = vec![0u8; ct.len() - TAG_BYTES]; + crypto_secretbox_open_easy(&mut plaintext, ct, &nonce, &key) + .map_err(|e| anyhow!("secretbox decrypt failed: {e}"))?; + Ok(String::from_utf8(plaintext)?) +} + +fn decode_base64_permissive(s: &str) -> Result, Error> { + for engine in [&URL_SAFE_NO_PAD, &URL_SAFE, &STANDARD_NO_PAD, &STANDARD] { + if let Ok(b) = engine.decode(s) { + return Ok(b); + } + } + Err(anyhow!("invalid base64")) +} diff --git a/crates/mirror/src/db.rs b/crates/mirror/src/db.rs new file mode 100644 index 00000000..704582cf --- /dev/null +++ b/crates/mirror/src/db.rs @@ -0,0 +1,118 @@ +//! DB helpers — connecting + loading rows. + +use anyhow::{Context, Error}; +use chrono::{DateTime, Utc}; +use sqlx::{postgres::PgPoolOptions, FromRow, Pool, Postgres}; +use std::{env, time::Duration}; + +use crate::Provider; + +pub async fn connect() -> Result, Error> { + let url = env::var("XATA_POSTGRES_URL").context("XATA_POSTGRES_URL not set")?; + let pool = PgPoolOptions::new() + .max_connections(8) + .min_connections(2) + .acquire_timeout(Duration::from_secs(12)) + .max_lifetime(Some(Duration::from_secs(60 * 14))) + .test_before_acquire(true) + .connect(&url) + .await?; + Ok(pool) +} + +/// Joined view of `mirror_sources` + user DID. +#[derive(Debug, Clone, FromRow)] +pub struct MirrorSourceRow { + pub user_id: String, + pub did: String, + pub provider: String, + pub enabled: bool, + pub external_username: Option, + pub encrypted_api_key: Option, + pub last_scrobble_seen_at: Option>, +} + +pub async fn load_enabled( + pool: &Pool, + provider: Provider, +) -> Result, Error> { + let rows = sqlx::query_as::<_, MirrorSourceRow>( + r#" + SELECT + m.user_id AS user_id, + u.did AS did, + m.provider AS provider, + m.enabled AS enabled, + m.external_username AS external_username, + m.encrypted_api_key AS encrypted_api_key, + m.last_scrobble_seen_at AS last_scrobble_seen_at + FROM mirror_sources m + JOIN users u ON u.xata_id = m.user_id + WHERE m.enabled = TRUE AND m.provider = $1 + "#, + ) + .bind(provider.as_str()) + .fetch_all(pool) + .await?; + Ok(rows) +} + +pub async fn load_one( + pool: &Pool, + user_id: &str, + provider: Provider, +) -> Result, Error> { + let row = sqlx::query_as::<_, MirrorSourceRow>( + r#" + SELECT + m.user_id AS user_id, + u.did AS did, + m.provider AS provider, + m.enabled AS enabled, + m.external_username AS external_username, + m.encrypted_api_key AS encrypted_api_key, + m.last_scrobble_seen_at AS last_scrobble_seen_at + FROM mirror_sources m + JOIN users u ON u.xata_id = m.user_id + WHERE m.user_id = $1 AND m.provider = $2 + LIMIT 1 + "#, + ) + .bind(user_id) + .bind(provider.as_str()) + .fetch_optional(pool) + .await?; + Ok(row) +} + +pub async fn user_id_for_did(pool: &Pool, did: &str) -> Result, Error> { + let row: Option<(String,)> = + sqlx::query_as(r#"SELECT xata_id FROM users WHERE did = $1 LIMIT 1"#) + .bind(did) + .fetch_optional(pool) + .await?; + Ok(row.map(|(id,)| id)) +} + +pub async fn touch_polled( + pool: &Pool, + user_id: &str, + provider: Provider, + last_scrobble_seen_at: Option>, +) -> Result<(), Error> { + sqlx::query( + r#" + UPDATE mirror_sources + SET last_polled_at = NOW(), + last_scrobble_seen_at = COALESCE($3, last_scrobble_seen_at), + xata_updatedat = NOW() + WHERE user_id = $1 AND provider = $2 + "#, + ) + .bind(user_id) + .bind(provider.as_str()) + .bind(last_scrobble_seen_at) + .execute(pool) + .await?; + Ok(()) +} diff --git a/crates/mirror/src/dedup.rs b/crates/mirror/src/dedup.rs new file mode 100644 index 00000000..998687ad --- /dev/null +++ b/crates/mirror/src/dedup.rs @@ -0,0 +1,56 @@ +//! ±120 second dedup against the `scrobbles` table. +//! +//! We don't want to mirror a play the user already has from another source — +//! e.g. a Spotify play that the spotify crate already scrobbled, then Last.fm +//! reports the same play 30 seconds later. + +use anyhow::Error; +use chrono::{DateTime, Utc}; +use sqlx::{Pool, Postgres}; +use tracing::debug; + +const WINDOW_SECS: i64 = 120; + +/// Returns true when an existing scrobble matches (user, title, artist) within +/// ±120s of `at`. Title and artist are compared case-insensitively against the +/// denormalized columns on the `tracks` table. +pub async fn already_scrobbled( + pool: &Pool, + user_id: &str, + title: &str, + artist: &str, + at: DateTime, +) -> Result { + let lo = at - chrono::Duration::seconds(WINDOW_SECS); + let hi = at + chrono::Duration::seconds(WINDOW_SECS); + + let row: Option<(i32,)> = sqlx::query_as( + r#" + SELECT 1 + FROM scrobbles s + JOIN tracks t ON t.xata_id = s.track_id + WHERE s.user_id = $1 + AND lower(t.title) = lower($2) + AND lower(t.artist) = lower($3) + AND s.timestamp BETWEEN $4 AND $5 + LIMIT 1 + "#, + ) + .bind(user_id) + .bind(title) + .bind(artist) + .bind(lo) + .bind(hi) + .fetch_optional(pool) + .await?; + + let hit = row.is_some(); + debug!( + user_id = %user_id, + title = %title, + artist = %artist, + hit, + "dedup check" + ); + Ok(hit) +} diff --git a/crates/mirror/src/lastfm.rs b/crates/mirror/src/lastfm.rs new file mode 100644 index 00000000..1b8fed4e --- /dev/null +++ b/crates/mirror/src/lastfm.rs @@ -0,0 +1,251 @@ +//! Last.fm mirror — `user.getRecentTracks` polled every 30 seconds per user. +//! +//! Last.fm returns the currently-playing track at the top of the response with +//! `@attr.nowplaying = "true"`; those have no `date` and we skip them. Only +//! items with `date.uts` strictly newer than the watermark are mirrored. + +use std::time::Duration; + +use anyhow::Error; +use chrono::{DateTime, Utc}; +use reqwest::Client; +use serde::Deserialize; +use sqlx::{Pool, Postgres}; +use tokio_util::sync::CancellationToken; +use tracing::{debug, error, info, warn}; + +use crate::{ + crypto, + db::{self, MirrorSourceRow}, + dedup, rocksky, + track::NormalizedTrack, + Provider, +}; + +const ENDPOINT: &str = "https://ws.audioscrobbler.com/2.0/"; +const POLL_INTERVAL: Duration = Duration::from_secs(30); +const RECENT_LIMIT: u32 = 50; + +pub async fn run_user( + pool: Pool, + http: Client, + row: MirrorSourceRow, + cancel: CancellationToken, +) -> Result<(), Error> { + let api_key = match row.encrypted_api_key.as_deref() { + Some(enc) => crypto::decrypt(enc)?, + None => { + warn!(user_id = %row.user_id, "Last.fm: no API key, exiting task"); + return Ok(()); + } + }; + let Some(username) = row.external_username.clone() else { + warn!(user_id = %row.user_id, "Last.fm: no external username, exiting task"); + return Ok(()); + }; + + info!( + user_id = %row.user_id, + did = %row.did, + username = %username, + interval_secs = POLL_INTERVAL.as_secs(), + "Last.fm: starting poll loop" + ); + + let mut watermark = row.last_scrobble_seen_at.unwrap_or_else(Utc::now); + + loop { + tokio::select! { + _ = cancel.cancelled() => { + info!(user_id = %row.user_id, "Last.fm: cancelled"); + return Ok(()); + } + _ = tokio::time::sleep(POLL_INTERVAL) => {} + } + + match poll_once(&pool, &http, &row, &api_key, &username, watermark).await { + Ok(new_watermark) => watermark = new_watermark, + Err(e) => { + error!(user_id = %row.user_id, error = %e, "Last.fm: poll failed"); + } + } + } +} + +async fn poll_once( + pool: &Pool, + http: &Client, + row: &MirrorSourceRow, + api_key: &str, + username: &str, + watermark: DateTime, +) -> Result, Error> { + debug!(user_id = %row.user_id, "Last.fm: polling getRecentTracks"); + + let resp: RecentTracksResponse = http + .get(ENDPOINT) + .query(&[ + ("method", "user.getrecenttracks"), + ("user", username), + ("api_key", api_key), + ("format", "json"), + ("limit", &RECENT_LIMIT.to_string()), + ("from", &watermark.timestamp().to_string()), + ]) + .send() + .await? + .error_for_status()? + .json() + .await?; + + let items = resp.recenttracks.track.unwrap_or_default(); + info!( + user_id = %row.user_id, + fetched = items.len(), + watermark = watermark.to_rfc3339(), + "Last.fm: fetched recent tracks" + ); + + let mut max_seen = watermark; + let mut mirrored = 0usize; + let mut skipped_nowplaying = 0usize; + let mut skipped_dedup = 0usize; + + // Last.fm returns newest-first; iterate oldest-first so the watermark + // moves monotonically. + for item in items.into_iter().rev() { + if item.attr.as_ref().and_then(|a| a.nowplaying.as_deref()) == Some("true") { + skipped_nowplaying += 1; + continue; + } + let Some(date) = item.date else { continue }; + let Ok(uts) = date.uts.parse::() else { + continue; + }; + let at = DateTime::from_timestamp(uts, 0).unwrap_or(watermark); + if at <= watermark { + continue; + } + + let title = item.name.trim().to_string(); + let artist = item.artist.text.unwrap_or_default().trim().to_string(); + let album = item.album.text.unwrap_or_default().trim().to_string(); + + if title.is_empty() || artist.is_empty() { + continue; + } + + if dedup::already_scrobbled(pool, &row.user_id, &title, &artist, at).await? { + info!( + user_id = %row.user_id, + title = %title, + artist = %artist, + at = at.to_rfc3339(), + "Last.fm: skipped (already scrobbled within 120s)" + ); + skipped_dedup += 1; + if at > max_seen { + max_seen = at; + } + continue; + } + + let track = NormalizedTrack { + title, + album_artist: artist.clone(), + artist, + album, + duration: 0, // user.getrecenttracks doesn't include duration + timestamp: uts, + mb_id: nonempty(item.mbid), + album_art: largest_image(&item.image), + spotify_link: None, + lastfm_link: nonempty(item.url), + }; + + info!( + user_id = %row.user_id, + title = %track.title, + artist = %track.artist, + at = at.to_rfc3339(), + "Last.fm: mirroring" + ); + match rocksky::create_scrobble(http, &row.did, &track, Provider::Lastfm).await { + Ok(_) => mirrored += 1, + Err(e) => warn!(user_id = %row.user_id, error = %e, "Last.fm: mirror failed"), + } + + if at > max_seen { + max_seen = at; + } + } + + info!( + user_id = %row.user_id, + mirrored, + skipped_dedup, + skipped_nowplaying, + "Last.fm: poll complete" + ); + + db::touch_polled(pool, &row.user_id, Provider::Lastfm, Some(max_seen)).await?; + Ok(max_seen) +} + +fn nonempty(s: Option) -> Option { + s.filter(|v| !v.trim().is_empty()) +} + +fn largest_image(images: &Option>) -> Option { + images + .as_ref()? + .iter() + .rev() + .find(|i| i.text.as_deref().is_some_and(|t| !t.trim().is_empty())) + .and_then(|i| i.text.clone()) +} + +#[derive(Debug, Deserialize)] +struct RecentTracksResponse { + recenttracks: RecentTracks, +} + +#[derive(Debug, Deserialize)] +struct RecentTracks { + track: Option>, +} + +#[derive(Debug, Deserialize)] +struct TrackItem { + name: String, + artist: TextField, + album: TextField, + url: Option, + mbid: Option, + image: Option>, + date: Option, + #[serde(rename = "@attr")] + attr: Option, +} + +#[derive(Debug, Deserialize)] +struct TextField { + #[serde(rename = "#text")] + text: Option, +} + +#[derive(Debug, Deserialize)] +struct DateField { + uts: String, +} + +#[derive(Debug, Deserialize)] +struct Image { + #[serde(rename = "#text")] + text: Option, +} + +#[derive(Debug, Deserialize)] +struct Attr { + nowplaying: Option, +} diff --git a/crates/mirror/src/lib.rs b/crates/mirror/src/lib.rs new file mode 100644 index 00000000..6fca115e --- /dev/null +++ b/crates/mirror/src/lib.rs @@ -0,0 +1,59 @@ +//! `rocksky-mirror` ingests plays from external scrobbling services +//! (Last.fm, ListenBrainz, Teal.fm) and re-publishes them into Rocksky via +//! xrpc `app.rocksky.scrobble.createScrobble`. +//! +//! Exposed as a rockskyd subcommand (`rockskyd mirror`); there is no +//! standalone binary. +//! +//! Shared plumbing: +//! * [`dedup`] — ±120s window check against the `scrobbles` table to avoid +//! mirroring a play the user already has from another source. +//! * [`rocksky`] — wraps the xrpc call. +//! * [`crypto`] — libsodium secretbox decrypt for the API keys persisted by +//! the API layer (`apps/api/src/lib/storage-crypto.ts`). + +pub mod crypto; +pub mod db; +pub mod dedup; +pub mod lastfm; +pub mod listenbrainz; +pub mod rocksky; +pub mod supervisor; +pub mod tealfm; +pub mod token; +pub mod track; + +pub const MIRROR_NATS_TOPIC: &str = "rocksky.mirror.user"; +pub const TEALFM_PLAY_NSID: &str = "fm.teal.alpha.feed.play"; + +/// Providers we mirror from. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum Provider { + Lastfm, + Listenbrainz, + Tealfm, +} + +impl Provider { + pub fn as_str(self) -> &'static str { + match self { + Provider::Lastfm => "lastfm", + Provider::Listenbrainz => "listenbrainz", + Provider::Tealfm => "tealfm", + } + } + + pub fn parse(s: &str) -> Option { + match s { + "lastfm" => Some(Provider::Lastfm), + "listenbrainz" => Some(Provider::Listenbrainz), + "tealfm" => Some(Provider::Tealfm), + _ => None, + } + } +} + +/// Entrypoint used by `rockskyd mirror`. +pub async fn run() -> anyhow::Result<()> { + supervisor::run().await +} diff --git a/crates/mirror/src/listenbrainz.rs b/crates/mirror/src/listenbrainz.rs new file mode 100644 index 00000000..eb4fc0d4 --- /dev/null +++ b/crates/mirror/src/listenbrainz.rs @@ -0,0 +1,226 @@ +//! ListenBrainz mirror — `user/{name}/listens?min_ts=...` polled every 30s. +//! +//! ListenBrainz tokens are sent as `Authorization: Token `. We page +//! with `min_ts` (exclusive lower bound) so the watermark trims the response +//! server-side. + +use std::time::Duration; + +use anyhow::Error; +use chrono::{DateTime, Utc}; +use reqwest::Client; +use serde::Deserialize; +use sqlx::{Pool, Postgres}; +use tokio_util::sync::CancellationToken; +use tracing::{debug, error, info, warn}; + +use crate::{ + crypto, + db::{self, MirrorSourceRow}, + dedup, rocksky, + track::NormalizedTrack, + Provider, +}; + +const POLL_INTERVAL: Duration = Duration::from_secs(30); +const RECENT_LIMIT: u32 = 50; + +pub async fn run_user( + pool: Pool, + http: Client, + row: MirrorSourceRow, + cancel: CancellationToken, +) -> Result<(), Error> { + // Token is optional for the read endpoint — public listens work without + // one — but we keep it for higher rate limits and parity with Last.fm. + let token = match row.encrypted_api_key.as_deref() { + Some(enc) => Some(crypto::decrypt(enc)?), + None => None, + }; + let Some(username) = row.external_username.clone() else { + warn!(user_id = %row.user_id, "ListenBrainz: no external username, exiting task"); + return Ok(()); + }; + + info!( + user_id = %row.user_id, + did = %row.did, + username = %username, + interval_secs = POLL_INTERVAL.as_secs(), + "ListenBrainz: starting poll loop" + ); + + let mut watermark = row.last_scrobble_seen_at.unwrap_or_else(Utc::now); + + loop { + tokio::select! { + _ = cancel.cancelled() => { + info!(user_id = %row.user_id, "ListenBrainz: cancelled"); + return Ok(()); + } + _ = tokio::time::sleep(POLL_INTERVAL) => {} + } + + match poll_once(&pool, &http, &row, token.as_deref(), &username, watermark).await { + Ok(new_watermark) => watermark = new_watermark, + Err(e) => { + error!(user_id = %row.user_id, error = %e, "ListenBrainz: poll failed"); + } + } + } +} + +async fn poll_once( + pool: &Pool, + http: &Client, + row: &MirrorSourceRow, + token: Option<&str>, + username: &str, + watermark: DateTime, +) -> Result, Error> { + debug!(user_id = %row.user_id, "ListenBrainz: polling listens"); + + let url = format!("https://api.listenbrainz.org/1/user/{username}/listens"); + let mut req = http.get(&url).query(&[ + ("count", &RECENT_LIMIT.to_string()), + ("min_ts", &watermark.timestamp().to_string()), + ]); + if let Some(t) = token { + req = req.header("Authorization", format!("Token {t}")); + } + + let resp: ListensResponse = req.send().await?.error_for_status()?.json().await?; + let items = resp.payload.listens.unwrap_or_default(); + + info!( + user_id = %row.user_id, + fetched = items.len(), + watermark = watermark.to_rfc3339(), + "ListenBrainz: fetched listens" + ); + + let mut max_seen = watermark; + let mut mirrored = 0usize; + let mut skipped_dedup = 0usize; + + // ListenBrainz returns newest-first; iterate oldest-first. + for listen in items.into_iter().rev() { + let at = DateTime::from_timestamp(listen.listened_at, 0).unwrap_or(watermark); + if at <= watermark { + continue; + } + + let m = &listen.track_metadata; + let title = m.track_name.trim().to_string(); + let artist = m.artist_name.trim().to_string(); + let album = m + .release_name + .clone() + .unwrap_or_default() + .trim() + .to_string(); + if title.is_empty() || artist.is_empty() { + continue; + } + + if dedup::already_scrobbled(pool, &row.user_id, &title, &artist, at).await? { + info!( + user_id = %row.user_id, + title = %title, + artist = %artist, + at = at.to_rfc3339(), + "ListenBrainz: skipped (already scrobbled within 120s)" + ); + skipped_dedup += 1; + if at > max_seen { + max_seen = at; + } + continue; + } + + let info = m.additional_info.as_ref(); + let duration_ms = info + .and_then(|i| i.duration_ms) + .or_else(|| info.and_then(|i| i.duration.map(|s| s as i64 * 1000))) + .unwrap_or(0); + + let track = NormalizedTrack { + title, + album_artist: artist.clone(), + artist, + album, + duration: duration_ms, + timestamp: listen.listened_at, + mb_id: m + .mbid_mapping + .as_ref() + .and_then(|m| m.recording_mbid.clone()), + album_art: None, + spotify_link: info.and_then(|i| i.spotify_id.clone()), + lastfm_link: None, + }; + + info!( + user_id = %row.user_id, + title = %track.title, + artist = %track.artist, + at = at.to_rfc3339(), + "ListenBrainz: mirroring" + ); + match rocksky::create_scrobble(http, &row.did, &track, Provider::Listenbrainz).await { + Ok(_) => mirrored += 1, + Err(e) => warn!(user_id = %row.user_id, error = %e, "ListenBrainz: mirror failed"), + } + + if at > max_seen { + max_seen = at; + } + } + + info!( + user_id = %row.user_id, + mirrored, + skipped_dedup, + "ListenBrainz: poll complete" + ); + + db::touch_polled(pool, &row.user_id, Provider::Listenbrainz, Some(max_seen)).await?; + Ok(max_seen) +} + +#[derive(Debug, Deserialize)] +struct ListensResponse { + payload: ListensPayload, +} + +#[derive(Debug, Deserialize)] +struct ListensPayload { + listens: Option>, +} + +#[derive(Debug, Deserialize)] +struct Listen { + listened_at: i64, + track_metadata: TrackMetadata, +} + +#[derive(Debug, Deserialize)] +struct TrackMetadata { + track_name: String, + artist_name: String, + release_name: Option, + additional_info: Option, + mbid_mapping: Option, +} + +#[derive(Debug, Deserialize)] +struct AdditionalInfo { + duration_ms: Option, + duration: Option, + spotify_id: Option, +} + +#[derive(Debug, Deserialize)] +struct MbidMapping { + recording_mbid: Option, +} diff --git a/crates/mirror/src/rocksky.rs b/crates/mirror/src/rocksky.rs new file mode 100644 index 00000000..c6c12be7 --- /dev/null +++ b/crates/mirror/src/rocksky.rs @@ -0,0 +1,66 @@ +//! XRPC client for `app.rocksky.scrobble.createScrobble`. +//! +//! Mirrors the call sites in `crates/spotify` and `crates/navidrome`: a JWT +//! bearer is built from the user's DID, then we POST the normalized track to +//! the API. The server handler is fire-and-forget (returns immediately after +//! kicking off the scrobble pipeline), but it does its own ±60s dedup, so a +//! duplicate slipping past our ±120s pre-check is still safe. + +use std::env; + +use anyhow::{Context, Error}; +use reqwest::Client; +use tracing::{info, warn}; + +use crate::{token, track::NormalizedTrack, Provider}; + +const DEFAULT_API: &str = "https://api.rocksky.app"; + +const CREATE_SCROBBLE_NSID: &str = "app.rocksky.scrobble.createScrobble"; + +pub async fn create_scrobble( + client: &Client, + did: &str, + track: &NormalizedTrack, + provider: Provider, +) -> Result<(), Error> { + let api = env::var("ROCKSKY_API").unwrap_or_else(|_| DEFAULT_API.to_string()); + let url = format!("{api}/xrpc/{CREATE_SCROBBLE_NSID}"); + + let bearer = token::generate(did).context("failed to mint JWT for createScrobble")?; + let res = client + .post(&url) + .bearer_auth(bearer) + .json(track) + .send() + .await?; + + let status = res.status(); + if !status.is_success() { + let body = res.text().await.unwrap_or_default(); + warn!( + provider = provider.as_str(), + %did, + title = %track.title, + artist = %track.artist, + %status, + %body, + "createScrobble returned non-success" + ); + return Err(anyhow::anyhow!( + "createScrobble {} returned {}", + url, + status + )); + } + + info!( + provider = provider.as_str(), + %did, + title = %track.title, + artist = %track.artist, + timestamp = track.timestamp, + "mirrored scrobble" + ); + Ok(()) +} diff --git a/crates/mirror/src/supervisor.rs b/crates/mirror/src/supervisor.rs new file mode 100644 index 00000000..1099c39a --- /dev/null +++ b/crates/mirror/src/supervisor.rs @@ -0,0 +1,209 @@ +//! Task supervisor: keeps one tokio task per enabled (provider, user) row for +//! Last.fm / ListenBrainz, and one shared task for Teal.fm. NATS messages on +//! [`crate::MIRROR_NATS_TOPIC`] (payload: `":"`) trigger +//! reconciliation for a single (provider, user_id). + +use std::collections::{HashMap, HashSet}; +use std::sync::Arc; +use std::time::Duration; + +use anyhow::{Context, Error}; +use async_nats::connect; +use futures_util::StreamExt; +use owo_colors::OwoColorize; +use reqwest::Client; +use sqlx::{Pool, Postgres}; +use tokio::sync::RwLock; +use tokio_util::sync::CancellationToken; +use tracing::{error, info, warn}; + +use crate::{db, lastfm, listenbrainz, tealfm, Provider, MIRROR_NATS_TOPIC}; + +type TaskMap = Arc>>; + +pub async fn run() -> Result<(), Error> { + let pool = db::connect().await.context("DB connect")?; + let http = Client::builder() + .timeout(Duration::from_secs(30)) + .user_agent("rocksky-mirror/0.1") + .build()?; + + let tasks: TaskMap = Arc::new(RwLock::new(HashMap::new())); + let tealfm_enabled: tealfm::EnabledDids = Arc::new(RwLock::new(HashSet::new())); + + // Initial reconcile: bring all currently enabled (provider, user_id) rows + // into a running state. + for provider in [Provider::Lastfm, Provider::Listenbrainz, Provider::Tealfm] { + let rows = db::load_enabled(&pool, provider).await?; + info!( + provider = provider.as_str(), + count = rows.len(), + "supervisor: hydrating enabled mirror sources" + ); + for row in rows { + spawn_for( + provider, + pool.clone(), + http.clone(), + row, + tasks.clone(), + tealfm_enabled.clone(), + ) + .await; + } + } + + // Always run the shared Teal.fm jetstream subscriber — it filters by the + // shared `tealfm_enabled` DID set, so an empty set just means we drop + // every event cheaply. + { + let pool = pool.clone(); + let http = http.clone(); + let enabled = tealfm_enabled.clone(); + let cancel = CancellationToken::new(); + tokio::spawn(async move { + if let Err(e) = tealfm::run(pool, http, enabled, cancel).await { + error!(error = %e, "Teal.fm: subscriber exited"); + } + }); + } + + // React to toggle events from the API. + let nats_url = std::env::var("NATS_URL").unwrap_or_else(|_| "nats://localhost:4222".into()); + let nc = connect(&nats_url) + .await + .with_context(|| format!("connecting to NATS at {nats_url}"))?; + info!(addr = %nats_url.bright_green(), "supervisor: connected to NATS"); + let mut sub = nc.subscribe(MIRROR_NATS_TOPIC.to_string()).await?; + info!(topic = %MIRROR_NATS_TOPIC.bright_green(), "supervisor: subscribed"); + + while let Some(msg) = sub.next().await { + let payload = match std::str::from_utf8(&msg.payload) { + Ok(s) => s.to_string(), + Err(e) => { + warn!(error = %e, "supervisor: non-utf8 NATS payload"); + continue; + } + }; + let Some((provider_str, user_id)) = payload.split_once(':') else { + warn!( + payload, + "supervisor: malformed payload, expected `provider:user_id`" + ); + continue; + }; + let Some(provider) = Provider::parse(provider_str) else { + warn!(payload, "supervisor: unknown provider"); + continue; + }; + + info!( + provider = provider.as_str(), + user_id, "supervisor: reconciling" + ); + if let Err(e) = reconcile( + provider, + user_id.to_string(), + pool.clone(), + http.clone(), + tasks.clone(), + tealfm_enabled.clone(), + ) + .await + { + error!(error = %e, "supervisor: reconcile failed"); + } + } + + Ok(()) +} + +async fn reconcile( + provider: Provider, + user_id: String, + pool: Pool, + http: Client, + tasks: TaskMap, + tealfm_enabled: tealfm::EnabledDids, +) -> Result<(), Error> { + let row = db::load_one(&pool, &user_id, provider).await?; + let key = (provider, user_id.clone()); + + match row { + Some(r) if r.enabled => { + // Restart-style reconcile: cancel any existing task before + // starting a fresh one, so updated credentials are picked up. + cancel_task(&tasks, &key).await; + spawn_for(provider, pool, http, r, tasks, tealfm_enabled).await; + } + _ => { + cancel_task(&tasks, &key).await; + if let Provider::Tealfm = provider { + if let Some(r) = row { + tealfm_enabled.write().await.remove(&r.did); + } + } + info!( + provider = provider.as_str(), + user_id, "supervisor: stopped (disabled or missing)" + ); + } + } + Ok(()) +} + +async fn cancel_task(tasks: &TaskMap, key: &(Provider, String)) { + if let Some(cancel) = tasks.write().await.remove(key) { + cancel.cancel(); + } +} + +async fn spawn_for( + provider: Provider, + pool: Pool, + http: Client, + row: db::MirrorSourceRow, + tasks: TaskMap, + tealfm_enabled: tealfm::EnabledDids, +) { + match provider { + Provider::Tealfm => { + // Teal.fm has no per-user task — we just register the DID with + // the shared subscriber. + tealfm_enabled.write().await.insert(row.did.clone()); + info!( + did = %row.did, + user_id = %row.user_id, + "Teal.fm: enabled DID registered" + ); + } + Provider::Lastfm | Provider::Listenbrainz => { + let cancel = CancellationToken::new(); + tasks + .write() + .await + .insert((provider, row.user_id.clone()), cancel.clone()); + + let pool_c = pool.clone(); + let http_c = http.clone(); + let user_id_c = row.user_id.clone(); + tokio::spawn(async move { + let result = match provider { + Provider::Lastfm => lastfm::run_user(pool_c, http_c, row, cancel).await, + Provider::Listenbrainz => { + listenbrainz::run_user(pool_c, http_c, row, cancel).await + } + Provider::Tealfm => unreachable!(), + }; + if let Err(e) = result { + error!( + provider = provider.as_str(), + user_id = %user_id_c, + error = %e, + "supervisor: per-user task exited with error" + ); + } + }); + } + } +} diff --git a/crates/mirror/src/tealfm.rs b/crates/mirror/src/tealfm.rs new file mode 100644 index 00000000..82525e98 --- /dev/null +++ b/crates/mirror/src/tealfm.rs @@ -0,0 +1,239 @@ +//! Teal.fm mirror — one Jetstream WebSocket connection for all enabled users. +//! +//! We subscribe to `fm.teal.alpha.feed.play` and look up the commit's `did` +//! against a per-process `enabled_dids` set. The Jetstream URL filter cheaply +//! drops everything outside this collection, so the only work we do per event +//! is a DID set lookup + optional dedup query. +//! +//! Note: the user spec said `fm.teal.alpha.play`, but the actual NSID used by +//! `apps/api/src/tealfm/index.ts` is `fm.teal.alpha.feed.play`. We use the +//! real one. + +use std::sync::Arc; +use std::time::Duration; + +use anyhow::Error; +use chrono::{DateTime, Utc}; +use futures_util::StreamExt; +use reqwest::Client; +use serde::Deserialize; +use sqlx::{Pool, Postgres}; +use std::collections::HashSet; +use std::env; +use tokio::sync::RwLock; +use tokio_tungstenite::{connect_async, tungstenite::Message}; +use tokio_util::sync::CancellationToken; +use tracing::{debug, error, info, warn}; + +use crate::{db, dedup, rocksky, track::NormalizedTrack, Provider, TEALFM_PLAY_NSID}; + +/// Set of DIDs currently mirroring Teal.fm — shared with the supervisor so it +/// can add/remove members in response to NATS toggle events. +pub type EnabledDids = Arc>>; + +pub async fn run( + pool: Pool, + http: Client, + enabled_dids: EnabledDids, + cancel: CancellationToken, +) -> Result<(), Error> { + let server = env::var("JETSTREAM_SERVER") + .unwrap_or_else(|_| "wss://jetstream2.us-west.bsky.network".to_string()); + let url = format!("{server}/subscribe?wantedCollections={TEALFM_PLAY_NSID}"); + + info!(url = %url, "Teal.fm: starting Jetstream subscriber"); + + loop { + if cancel.is_cancelled() { + info!("Teal.fm: cancelled"); + return Ok(()); + } + + match run_once(&pool, &http, &enabled_dids, &url, &cancel).await { + Ok(_) => warn!("Teal.fm: subscriber returned cleanly, reconnecting"), + Err(e) => error!(error = %e, "Teal.fm: subscriber error"), + } + tokio::time::sleep(Duration::from_secs(2)).await; + } +} + +async fn run_once( + pool: &Pool, + http: &Client, + enabled_dids: &EnabledDids, + url: &str, + cancel: &CancellationToken, +) -> Result<(), Error> { + let (mut ws, _) = connect_async(url).await?; + info!(url = %url, "Teal.fm: connected to Jetstream"); + + loop { + tokio::select! { + _ = cancel.cancelled() => return Ok(()), + msg = ws.next() => { + let Some(msg) = msg else { + warn!("Teal.fm: stream ended"); + return Ok(()); + }; + match msg { + Ok(Message::Text(text)) => { + if let Err(e) = handle_event(pool, http, enabled_dids, &text).await { + warn!(error = %e, "Teal.fm: handle_event failed"); + } + } + Ok(Message::Ping(_)) | Ok(Message::Pong(_)) | Ok(Message::Binary(_)) => {} + Ok(Message::Close(_)) => { + info!("Teal.fm: Jetstream closed connection"); + return Ok(()); + } + Ok(Message::Frame(_)) => {} + Err(e) => { + warn!(error = %e, "Teal.fm: WS error"); + return Ok(()); + } + } + } + } + } +} + +async fn handle_event( + pool: &Pool, + http: &Client, + enabled_dids: &EnabledDids, + text: &str, +) -> Result<(), Error> { + let evt: JetstreamEvent = serde_json::from_str(text)?; + if evt.kind != "commit" { + return Ok(()); + } + let Some(commit) = evt.commit else { + return Ok(()); + }; + if commit.collection != TEALFM_PLAY_NSID { + return Ok(()); + } + if commit.operation != "create" { + debug!(operation = %commit.operation, "Teal.fm: ignoring non-create"); + return Ok(()); + } + + // Cheap path first: is this DID enabled? + let did = evt.did; + { + let set = enabled_dids.read().await; + if !set.contains(&did) { + return Ok(()); + } + } + + let Some(record) = commit.record else { + return Ok(()); + }; + let play: PlayRecord = match serde_json::from_value(record) { + Ok(p) => p, + Err(e) => { + warn!(error = %e, "Teal.fm: failed to parse play record"); + return Ok(()); + } + }; + + let Some(user_id) = db::user_id_for_did(pool, &did).await? else { + warn!(did = %did, "Teal.fm: enabled DID has no users row, dropping from set"); + enabled_dids.write().await.remove(&did); + return Ok(()); + }; + + let title = play.track_name.trim().to_string(); + let artist = play + .artists + .first() + .map(|a| a.artist_name.trim().to_string()) + .unwrap_or_default(); + let album = play.release_name.unwrap_or_default().trim().to_string(); + if title.is_empty() || artist.is_empty() { + return Ok(()); + } + + let at: DateTime = play + .played_time + .parse::>() + .unwrap_or_else(|_| Utc::now()); + + info!( + did = %did, + user_id = %user_id, + title = %title, + artist = %artist, + at = at.to_rfc3339(), + "Teal.fm: received play event" + ); + + if dedup::already_scrobbled(pool, &user_id, &title, &artist, at).await? { + info!( + did = %did, + title = %title, + artist = %artist, + "Teal.fm: skipped (already scrobbled within 120s)" + ); + return Ok(()); + } + + let track = NormalizedTrack { + title, + album_artist: artist.clone(), + artist, + album, + duration: play.duration.map(|d| d as i64 * 1000).unwrap_or(0), + timestamp: at.timestamp(), + mb_id: play.recording_mb_id.map(strip_mbid_prefix), + album_art: None, + spotify_link: None, + lastfm_link: None, + }; + + info!( + did = %did, + title = %track.title, + "Teal.fm: mirroring" + ); + if let Err(e) = rocksky::create_scrobble(http, &did, &track, Provider::Tealfm).await { + warn!(did = %did, error = %e, "Teal.fm: mirror failed"); + } + Ok(()) +} + +fn strip_mbid_prefix(s: String) -> String { + s.strip_prefix("mbid:").map(str::to_string).unwrap_or(s) +} + +#[derive(Debug, Deserialize)] +struct JetstreamEvent { + did: String, + kind: String, + commit: Option, +} + +#[derive(Debug, Deserialize)] +struct Commit { + operation: String, + collection: String, + record: Option, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct PlayRecord { + track_name: String, + played_time: String, + duration: Option, + release_name: Option, + artists: Vec, + recording_mb_id: Option, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +struct PlayArtist { + artist_name: String, +} diff --git a/crates/mirror/src/token.rs b/crates/mirror/src/token.rs new file mode 100644 index 00000000..077d67ef --- /dev/null +++ b/crates/mirror/src/token.rs @@ -0,0 +1,29 @@ +//! JWT bearer used by the xrpc client. Mirrors `crates/spotify/src/token.rs`. + +use std::env; + +use anyhow::{Context, Error}; +use jsonwebtoken::{encode, EncodingKey, Header}; +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Serialize, Deserialize)] +struct Claims { + exp: usize, + iat: usize, + did: String, +} + +pub fn generate(did: &str) -> Result { + let secret = env::var("JWT_SECRET").context("JWT_SECRET not set")?; + let now = chrono::Utc::now().timestamp() as usize; + let claims = Claims { + exp: now + 3600, + iat: now, + did: did.to_string(), + }; + Ok(encode( + &Header::default(), + &claims, + &EncodingKey::from_secret(secret.as_ref()), + )?) +} diff --git a/crates/mirror/src/track.rs b/crates/mirror/src/track.rs new file mode 100644 index 00000000..e0e7100a --- /dev/null +++ b/crates/mirror/src/track.rs @@ -0,0 +1,38 @@ +//! Normalized track payload posted to `app.rocksky.scrobble.createScrobble`. +//! +//! Matches the Zod schema at `apps/api/src/types/track.ts`. Only the four +//! required fields are mandatory; everything else is best-effort enrichment +//! from the source's response. + +use chrono::{DateTime, Utc}; +use serde::Serialize; + +#[derive(Debug, Clone, Serialize)] +pub struct NormalizedTrack { + pub title: String, + pub artist: String, + pub album: String, + #[serde(rename = "albumArtist")] + pub album_artist: String, + + /// Track duration in **milliseconds** (matches the Track schema). + pub duration: i64, + + /// Unix seconds for when the play occurred (Track schema allows this). + pub timestamp: i64, + + #[serde(skip_serializing_if = "Option::is_none", rename = "mbId")] + pub mb_id: Option, + #[serde(skip_serializing_if = "Option::is_none", rename = "albumArt")] + pub album_art: Option, + #[serde(skip_serializing_if = "Option::is_none", rename = "spotifyLink")] + pub spotify_link: Option, + #[serde(skip_serializing_if = "Option::is_none", rename = "lastfmLink")] + pub lastfm_link: Option, +} + +impl NormalizedTrack { + pub fn at(&self) -> DateTime { + DateTime::from_timestamp(self.timestamp, 0).unwrap_or_else(Utc::now) + } +} diff --git a/crates/rockskyd/Cargo.toml b/crates/rockskyd/Cargo.toml index 51b3b04c..50fbe20e 100644 --- a/crates/rockskyd/Cargo.toml +++ b/crates/rockskyd/Cargo.toml @@ -14,6 +14,7 @@ anyhow = "1.0.96" rocksky-dropbox = { path = "../dropbox" } rocksky-googledrive = { path = "../googledrive" } rocksky-jetstream = { path = "../jetstream" } +rocksky-mirror = { path = "../mirror" } rocksky-playlists = { path = "../playlists" } rocksky-scrobbler = { path = "../scrobbler" } rocksky-spotify = { path = "../spotify" } diff --git a/crates/rockskyd/src/cmd/mirror.rs b/crates/rockskyd/src/cmd/mirror.rs new file mode 100644 index 00000000..664a058f --- /dev/null +++ b/crates/rockskyd/src/cmd/mirror.rs @@ -0,0 +1,6 @@ +use anyhow::Error; + +pub async fn start_mirror_service() -> Result<(), Error> { + rocksky_mirror::run().await?; + Ok(()) +} diff --git a/crates/rockskyd/src/cmd/mod.rs b/crates/rockskyd/src/cmd/mod.rs index d48cf4b7..817d3ac9 100644 --- a/crates/rockskyd/src/cmd/mod.rs +++ b/crates/rockskyd/src/cmd/mod.rs @@ -1,6 +1,7 @@ pub mod dropbox; pub mod googledrive; pub mod jetstream; +pub mod mirror; pub mod navidrome; pub mod playlist; pub mod pull; diff --git a/crates/rockskyd/src/main.rs b/crates/rockskyd/src/main.rs index 1cc3bbef..60a00c5d 100644 --- a/crates/rockskyd/src/main.rs +++ b/crates/rockskyd/src/main.rs @@ -21,6 +21,7 @@ fn cli() -> Command { .subcommand(Command::new("serve").about("Serve Rocksky Google Drive API")), ) .subcommand(Command::new("jetstream").about("Start JetStream Subscriber Service")) + .subcommand(Command::new("mirror").about("Mirror plays from Last.fm, ListenBrainz, Teal.fm into Rocksky")) .subcommand(Command::new("navidrome").about("Start Navidrome-compatible API (Subsonic REST API)")) .subcommand(Command::new("playlist").about("Playlist related commands")) .subcommand(Command::new("scrobbler").about("Start Scrobbler API")) @@ -64,6 +65,9 @@ async fn main() -> Result<(), Box> { Some(("jetstream", _)) => { cmd::jetstream::start_jetstream_service().await?; } + Some(("mirror", _)) => { + cmd::mirror::start_mirror_service().await?; + } Some(("navidrome", _)) => { cmd::navidrome::start_navidrome_service().await?; }