rocksky.app
A decentralized music tracking and discovery platform built on AT Protocol 🎵
rocksky.app
spotify
atproto
lastfm
musicbrainz
scrobbling
listenbrainz
fix(scrobbles): prevent duplicate scrobbles across sources
Same track+timestamp from different at-uris was slipping past the
uri-only unique constraint. Three layers:
- DB UNIQUE (user_id, track_id, timestamp) + dedupe migration that
re-points shouts before deleting losing rows.
- ON CONFLICT DO NOTHING in jetstream + navidrome inserts; jetstream
skips the NATS publish + Discord webhook on conflict.
- Redis SETNX guard around putScrobbleRecord so concurrent sources
(Spotify/Last.fm/Navidrome) can't each write their own at-record
for the same listen.
author
Tsiry Sandratraina
date
(Jun 9, 2026, 6:07 PM +0300)
commit
8621b616
8621b616ddc79b4554fd7698a63093e1c132db5c
parent
09e9990f
09e9990fbcffd21bd441666a39527c7c5013c42a
+112
-17
+47
apps/api/drizzle/0014_scrobbles_unique_user_track_ts.sql
+47
apps/api/drizzle/0014_scrobbles_unique_user_track_ts.sql
···
1
+
-- Same (user_id, track_id, timestamp) was being inserted multiple times from
2
+
-- different ingestion sources (jetstream/navidrome/mirrors); each produced a
3
+
-- distinct at-uri so scrobbles_uri_unique didn't catch it. Backfill: collapse
4
+
-- existing duplicates (keep earliest by xata_createdat), then enforce a
5
+
-- composite unique so future races short-circuit at the DB.
6
+
7
+
-- Re-point shouts attached to a soon-to-be-deleted duplicate at the survivor.
8
+
WITH ranked AS (
9
+
SELECT
10
+
xata_id,
11
+
user_id,
12
+
track_id,
13
+
"timestamp",
14
+
FIRST_VALUE(xata_id) OVER (
15
+
PARTITION BY user_id, track_id, "timestamp"
16
+
ORDER BY xata_createdat ASC, xata_id ASC
17
+
) AS keep_id
18
+
FROM scrobbles
19
+
WHERE user_id IS NOT NULL AND track_id IS NOT NULL
20
+
)
21
+
UPDATE shouts s
22
+
SET scrobble_id = r.keep_id
23
+
FROM ranked r
24
+
WHERE s.scrobble_id = r.xata_id
25
+
AND r.xata_id <> r.keep_id;
26
+
--> statement-breakpoint
27
+
28
+
-- Drop every duplicate row that isn't the survivor.
29
+
WITH ranked AS (
30
+
SELECT
31
+
xata_id,
32
+
FIRST_VALUE(xata_id) OVER (
33
+
PARTITION BY user_id, track_id, "timestamp"
34
+
ORDER BY xata_createdat ASC, xata_id ASC
35
+
) AS keep_id
36
+
FROM scrobbles
37
+
WHERE user_id IS NOT NULL AND track_id IS NOT NULL
38
+
)
39
+
DELETE FROM scrobbles
40
+
WHERE xata_id IN (
41
+
SELECT xata_id FROM ranked WHERE xata_id <> keep_id
42
+
);
43
+
--> statement-breakpoint
44
+
45
+
ALTER TABLE "scrobbles"
46
+
ADD CONSTRAINT "scrobbles_user_track_timestamp_unique"
47
+
UNIQUE ("user_id", "track_id", "timestamp");
+7
apps/api/drizzle/meta/_journal.json
+7
apps/api/drizzle/meta/_journal.json
+23
apps/api/src/nowplaying/nowplaying.service.ts
+23
apps/api/src/nowplaying/nowplaying.service.ts
···
1059
1059
if (!track.albumArt) track.albumArt = existingAlbum.albumArt;
1060
1060
}
1061
1061
1062
+
// Cross-process lock: when multiple sources (Spotify webhook, Last.fm mirror,
1063
+
// Navidrome, etc.) fire for the same listen within the same second, they all
1064
+
// arrive here with the same userDid + track + timestamp and would each call
1065
+
// putRecord, leaving duplicate at://app.rocksky.scrobble records on the user's
1066
+
// PDS. The DB unique on (user, track, timestamp) catches dupes at storage, but
1067
+
// by then the redundant at-records already exist. SETNX on Redis ensures only
1068
+
// the first caller writes the at-record; the rest exit before the put. Lock
1069
+
// identity uses lowercased title+artist (the values the 60s window already
1070
+
// dedupes against) plus the exact integer-second timestamp.
1071
+
const lockTs = track.timestamp || dayjs().unix();
1072
+
const lockHash = createHash("sha256")
1073
+
.update(`${track.title.toLowerCase()}|${track.artist.toLowerCase()}`)
1074
+
.digest("hex");
1075
+
const lockKey = `scrobble-put:${userDid}:${lockHash}:${lockTs}`;
1076
+
const lockAcquired = await ctx.redis.set(lockKey, "1", { NX: true, EX: 120 });
1077
+
if (lockAcquired !== "OK") {
1078
+
consola.info(
1079
+
`Scrobble put lock held by concurrent source for ${chalk.cyan(track.title)} @ ${chalk.cyan(
1080
+
dayjs.unix(lockTs).format("YYYY-MM-DD HH:mm:ss"),
1081
+
)} — skipping putRecord`,
1082
+
);
1083
+
return;
1084
+
}
1062
1085
const scrobbleUri = await putScrobbleRecord(track, agent);
1063
1086
1064
1087
// loop while scrobble is null, try 30 times, sleep 1 second between tries
+13
-1
apps/api/src/schema/scrobbles.ts
+13
-1
apps/api/src/schema/scrobbles.ts
···
1
1
import { type InferInsertModel, type InferSelectModel, sql } from "drizzle-orm";
2
-
import { index, integer, pgTable, text, timestamp } from "drizzle-orm/pg-core";
2
+
import {
3
+
index,
4
+
integer,
5
+
pgTable,
6
+
text,
7
+
timestamp,
8
+
unique,
9
+
} from "drizzle-orm/pg-core";
3
10
import albums from "./albums";
4
11
import artists from "./artists";
5
12
import tracks from "./tracks";
···
25
32
index("scrobbles_album_id_idx").on(t.albumId),
26
33
index("scrobbles_track_id_idx").on(t.trackId),
27
34
index("scrobbles_timestamp_idx").on(t.timestamp),
35
+
unique("scrobbles_user_track_timestamp_unique").on(
36
+
t.userId,
37
+
t.trackId,
38
+
t.timestamp,
39
+
),
28
40
],
29
41
);
30
42
+2
-14
apps/api/src/scripts/collections.ts
+2
-14
apps/api/src/scripts/collections.ts
···
193
193
194
194
try {
195
195
if (!newScrobble) {
196
+
// Conflict path — either the at-uri or the
197
+
// (user_id, track_id, timestamp) composite already exists.
196
198
[newScrobble] = await ctx.db
197
199
.select()
198
200
.from(schema.scrobbles)
···
201
203
and(
202
204
eq(schema.scrobbles.userId, user.id),
203
205
eq(schema.scrobbles.trackId, track.id),
204
-
eq(schema.scrobbles.artistId, artist.id),
205
206
eq(schema.scrobbles.timestamp, new Date(value.createdAt)),
206
207
),
207
208
eq(schema.scrobbles.uri, scrobble.uri),
···
216
217
scrobble.uri,
217
218
)} — skipping publish`,
218
219
);
219
-
await ctx.db
220
-
.insert(schema.scrobbles)
221
-
.values({
222
-
albumId: album.id,
223
-
trackId: track.id,
224
-
artistId: artist.id,
225
-
uri: scrobble.uri,
226
-
userId: user.id,
227
-
timestamp: new Date(value.createdAt),
228
-
createdAt: new Date(value.createdAt),
229
-
})
230
-
.returning()
231
-
.execute();
232
220
return;
233
221
}
234
222
await publishScrobble(ctx, newScrobble.id);
+19
-2
crates/jetstream/src/repo.rs
+19
-2
crates/jetstream/src/repo.rs
···
82
82
83
83
tracing::info!(title = %scrobble_record.title.magenta(), artist = %scrobble_record.artist.magenta(), album = %scrobble_record.album.magenta(), "Saving scrobble");
84
84
85
-
let scrobble_id: String = sqlx::query_scalar(
85
+
// ON CONFLICT on the (user, track, timestamp) composite drops the
86
+
// race between concurrent sources (Spotify webhook, Last.fm mirror,
87
+
// Navidrome, etc.) that each publish their own at-uri for the same
88
+
// listen. RETURNING is empty on conflict — that's the signal to
89
+
// skip the downstream fan-out (NATS + Discord) so we don't double-
90
+
// fire for what is, by definition, the same scrobble.
91
+
let scrobble_id: Option<String> = sqlx::query_scalar(
86
92
r#"
87
93
INSERT INTO scrobbles (
88
94
album_id,
···
92
98
user_id,
93
99
timestamp
94
100
) VALUES ($1, $2, $3, $4, $5, $6)
101
+
ON CONFLICT (user_id, track_id, timestamp) DO NOTHING
95
102
RETURNING xata_id
96
103
"#,
97
104
)
···
105
112
.unwrap()
106
113
.with_timezone(&chrono::Utc),
107
114
)
108
-
.fetch_one(&mut *tx)
115
+
.fetch_optional(&mut *tx)
109
116
.await?;
110
117
111
118
tx.commit().await?;
119
+
120
+
let Some(scrobble_id) = scrobble_id else {
121
+
tracing::info!(
122
+
title = %scrobble_record.title.magenta(),
123
+
artist = %scrobble_record.artist.magenta(),
124
+
did = %did,
125
+
"Duplicate scrobble (same user/track/timestamp) — skipping publish"
126
+
);
127
+
return Ok(());
128
+
};
112
129
113
130
nc.publish("rocksky.scrobble.new", scrobble_id.into())
114
131
.await?;