From 7c9c93b4c6bec596487f999fc4c0e3285067358a Mon Sep 17 00:00:00 2001 From: User Date: Mon, 17 Aug 2026 15:34:25 -0700 Subject: [PATCH 01/14] feat: persist subscription group records Add account-scoped group and membership tables plus the API models used by the group service. Register both tables in production and test database setup. --- .../dev/typetype/server/db/DatabaseFactory.kt | 4 ++++ .../SubscriptionGroupMembershipsTable.kt | 17 +++++++++++++++++ .../db/tables/SubscriptionGroupsTable.kt | 18 ++++++++++++++++++ .../server/models/SubscriptionGroupItem.kt | 12 ++++++++++++ .../SubscriptionGroupMembershipRequest.kt | 6 ++++++ .../server/models/SubscriptionGroupRequest.kt | 6 ++++++ .../services/SubscriptionGroupResults.kt | 17 +++++++++++++++++ .../kotlin/dev/typetype/server/TestDatabase.kt | 4 ++++ 8 files changed, 84 insertions(+) create mode 100644 src/main/kotlin/dev/typetype/server/db/tables/SubscriptionGroupMembershipsTable.kt create mode 100644 src/main/kotlin/dev/typetype/server/db/tables/SubscriptionGroupsTable.kt create mode 100644 src/main/kotlin/dev/typetype/server/models/SubscriptionGroupItem.kt create mode 100644 src/main/kotlin/dev/typetype/server/models/SubscriptionGroupMembershipRequest.kt create mode 100644 src/main/kotlin/dev/typetype/server/models/SubscriptionGroupRequest.kt create mode 100644 src/main/kotlin/dev/typetype/server/services/SubscriptionGroupResults.kt diff --git a/src/main/kotlin/dev/typetype/server/db/DatabaseFactory.kt b/src/main/kotlin/dev/typetype/server/db/DatabaseFactory.kt index 0da240a5..4d2eeb7e 100644 --- a/src/main/kotlin/dev/typetype/server/db/DatabaseFactory.kt +++ b/src/main/kotlin/dev/typetype/server/db/DatabaseFactory.kt @@ -16,6 +16,8 @@ import dev.typetype.server.db.tables.SearchHistoryTable import dev.typetype.server.db.tables.SettingsTable import dev.typetype.server.db.tables.SessionsTable import dev.typetype.server.db.tables.SubscriptionsTable +import dev.typetype.server.db.tables.SubscriptionGroupMembershipsTable +import dev.typetype.server.db.tables.SubscriptionGroupsTable import dev.typetype.server.db.tables.UsersTable import dev.typetype.server.db.tables.UserAvatarsTable import dev.typetype.server.db.tables.WatchLaterTable @@ -57,6 +59,8 @@ object DatabaseFactory { AdminSettingsTable, HistoryTable, SubscriptionsTable, + SubscriptionGroupsTable, + SubscriptionGroupMembershipsTable, PlaylistsTable, PlaylistVideosTable, WatchLaterTable, diff --git a/src/main/kotlin/dev/typetype/server/db/tables/SubscriptionGroupMembershipsTable.kt b/src/main/kotlin/dev/typetype/server/db/tables/SubscriptionGroupMembershipsTable.kt new file mode 100644 index 00000000..93931a89 --- /dev/null +++ b/src/main/kotlin/dev/typetype/server/db/tables/SubscriptionGroupMembershipsTable.kt @@ -0,0 +1,17 @@ +package dev.typetype.server.db.tables + +import org.jetbrains.exposed.v1.core.ReferenceOption +import org.jetbrains.exposed.v1.core.Table + +object SubscriptionGroupMembershipsTable : Table("subscription_group_memberships") { + val groupId = text("group_id").references(SubscriptionGroupsTable.id, onDelete = ReferenceOption.CASCADE) + val userId = text("user_id") + val channelUrl = text("channel_url") + val addedAt = long("added_at") + + init { + index(false, userId, channelUrl) + } + + override val primaryKey = PrimaryKey(groupId, channelUrl) +} diff --git a/src/main/kotlin/dev/typetype/server/db/tables/SubscriptionGroupsTable.kt b/src/main/kotlin/dev/typetype/server/db/tables/SubscriptionGroupsTable.kt new file mode 100644 index 00000000..13b89b20 --- /dev/null +++ b/src/main/kotlin/dev/typetype/server/db/tables/SubscriptionGroupsTable.kt @@ -0,0 +1,18 @@ +package dev.typetype.server.db.tables + +import org.jetbrains.exposed.v1.core.Table + +object SubscriptionGroupsTable : Table("subscription_groups") { + val id = text("id") + val userId = text("user_id") + val name = text("name") + val normalizedName = text("normalized_name") + val createdAt = long("created_at") + val updatedAt = long("updated_at") + + init { + uniqueIndex(userId, normalizedName) + } + + override val primaryKey = PrimaryKey(id) +} diff --git a/src/main/kotlin/dev/typetype/server/models/SubscriptionGroupItem.kt b/src/main/kotlin/dev/typetype/server/models/SubscriptionGroupItem.kt new file mode 100644 index 00000000..684382bb --- /dev/null +++ b/src/main/kotlin/dev/typetype/server/models/SubscriptionGroupItem.kt @@ -0,0 +1,12 @@ +package dev.typetype.server.models + +import kotlinx.serialization.Serializable + +@Serializable +data class SubscriptionGroupItem( + val id: String, + val name: String, + val channelCount: Int, + val createdAt: Long, + val updatedAt: Long, +) diff --git a/src/main/kotlin/dev/typetype/server/models/SubscriptionGroupMembershipRequest.kt b/src/main/kotlin/dev/typetype/server/models/SubscriptionGroupMembershipRequest.kt new file mode 100644 index 00000000..9a2fcb5f --- /dev/null +++ b/src/main/kotlin/dev/typetype/server/models/SubscriptionGroupMembershipRequest.kt @@ -0,0 +1,6 @@ +package dev.typetype.server.models + +import kotlinx.serialization.Serializable + +@Serializable +data class SubscriptionGroupMembershipRequest(val channelUrl: String) diff --git a/src/main/kotlin/dev/typetype/server/models/SubscriptionGroupRequest.kt b/src/main/kotlin/dev/typetype/server/models/SubscriptionGroupRequest.kt new file mode 100644 index 00000000..136b831c --- /dev/null +++ b/src/main/kotlin/dev/typetype/server/models/SubscriptionGroupRequest.kt @@ -0,0 +1,6 @@ +package dev.typetype.server.models + +import kotlinx.serialization.Serializable + +@Serializable +data class SubscriptionGroupRequest(val name: String) diff --git a/src/main/kotlin/dev/typetype/server/services/SubscriptionGroupResults.kt b/src/main/kotlin/dev/typetype/server/services/SubscriptionGroupResults.kt new file mode 100644 index 00000000..6c19f560 --- /dev/null +++ b/src/main/kotlin/dev/typetype/server/services/SubscriptionGroupResults.kt @@ -0,0 +1,17 @@ +package dev.typetype.server.services + +import dev.typetype.server.models.SubscriptionGroupItem + +sealed interface SubscriptionGroupWriteResult { + data class Success(val group: SubscriptionGroupItem) : SubscriptionGroupWriteResult + data object InvalidName : SubscriptionGroupWriteResult + data object DuplicateName : SubscriptionGroupWriteResult + data object NotFound : SubscriptionGroupWriteResult +} + +sealed interface SubscriptionGroupMembershipResult { + data object Success : SubscriptionGroupMembershipResult + data object GroupNotFound : SubscriptionGroupMembershipResult + data object SubscriptionNotFound : SubscriptionGroupMembershipResult + data object MembershipNotFound : SubscriptionGroupMembershipResult +} diff --git a/src/test/kotlin/dev/typetype/server/TestDatabase.kt b/src/test/kotlin/dev/typetype/server/TestDatabase.kt index 1a22cc72..be1dc356 100644 --- a/src/test/kotlin/dev/typetype/server/TestDatabase.kt +++ b/src/test/kotlin/dev/typetype/server/TestDatabase.kt @@ -24,6 +24,8 @@ import dev.typetype.server.db.tables.SearchHistoryTable import dev.typetype.server.db.tables.SettingsTable import dev.typetype.server.db.tables.SessionsTable import dev.typetype.server.db.tables.SubscriptionsTable +import dev.typetype.server.db.tables.SubscriptionGroupMembershipsTable +import dev.typetype.server.db.tables.SubscriptionGroupsTable import dev.typetype.server.db.tables.YoutubeTakeoutImportJobsTable import dev.typetype.server.db.tables.YoutubeTakeoutPlaylistKeysTable import dev.typetype.server.db.tables.YoutubeSessionPairingsTable @@ -102,6 +104,8 @@ object TestDatabase { HistoryTable.deleteAll() FavoritesTable.deleteAll() SettingsTable.deleteAll() + SubscriptionGroupMembershipsTable.deleteAll() + SubscriptionGroupsTable.deleteAll() SubscriptionsTable.deleteAll() WatchLaterTable.deleteAll() ProgressTable.deleteAll() From 7ef9808753ce7cf464dd4f2336464b7e5f598083 Mon Sep 17 00:00:00 2001 From: User Date: Mon, 17 Aug 2026 15:34:33 -0700 Subject: [PATCH 02/14] feat: manage subscription group membership Create, rename, and remove account-owned groups and assign subscribed channels to multiple groups. Keep memberships consistent when subscriptions are deleted or replaced by imports. --- .../PipePipeBackupPersisterService.kt | 1 + .../SubscriptionGroupMembershipCleaner.kt | 19 ++ .../services/SubscriptionGroupsService.kt | 182 ++++++++++++++++++ .../server/services/SubscriptionSelection.kt | 17 ++ .../server/services/SubscriptionsService.kt | 36 +++- .../services/TypeTypeBackupCoreRestore.kt | 1 + 6 files changed, 255 insertions(+), 1 deletion(-) create mode 100644 src/main/kotlin/dev/typetype/server/services/SubscriptionGroupMembershipCleaner.kt create mode 100644 src/main/kotlin/dev/typetype/server/services/SubscriptionGroupsService.kt create mode 100644 src/main/kotlin/dev/typetype/server/services/SubscriptionSelection.kt diff --git a/src/main/kotlin/dev/typetype/server/services/PipePipeBackupPersisterService.kt b/src/main/kotlin/dev/typetype/server/services/PipePipeBackupPersisterService.kt index 9378fb54..4e965014 100644 --- a/src/main/kotlin/dev/typetype/server/services/PipePipeBackupPersisterService.kt +++ b/src/main/kotlin/dev/typetype/server/services/PipePipeBackupPersisterService.kt @@ -22,6 +22,7 @@ class PipePipeBackupPersisterService { .toMap() val history = insertHistory(userId, snapshot.history, avatarsByChannel) val subscriptions = insertSubscriptions(userId, snapshot.subscriptions) + SubscriptionGroupMembershipCleaner.retain(userId, snapshot.subscriptions.map { it.url }) val (playlists, playlistVideos) = insertPlaylists(userId, snapshot.playlists) val progress = insertProgress(userId, snapshot.progress) val searchHistory = insertSearchHistory(userId, snapshot.searchHistory) diff --git a/src/main/kotlin/dev/typetype/server/services/SubscriptionGroupMembershipCleaner.kt b/src/main/kotlin/dev/typetype/server/services/SubscriptionGroupMembershipCleaner.kt new file mode 100644 index 00000000..d78f7833 --- /dev/null +++ b/src/main/kotlin/dev/typetype/server/services/SubscriptionGroupMembershipCleaner.kt @@ -0,0 +1,19 @@ +package dev.typetype.server.services + +import dev.typetype.server.db.tables.SubscriptionGroupMembershipsTable +import org.jetbrains.exposed.v1.core.and +import org.jetbrains.exposed.v1.core.eq +import org.jetbrains.exposed.v1.core.notInList +import org.jetbrains.exposed.v1.jdbc.deleteWhere + +internal object SubscriptionGroupMembershipCleaner { + fun retain(userId: String, channelUrls: Collection) { + val retained = channelUrls.mapTo(linkedSetOf(), ChannelUrlCanonicalizer::canonicalize) + SubscriptionGroupMembershipsTable.deleteWhere { + val ownedByUser = SubscriptionGroupMembershipsTable.userId eq userId + if (retained.isEmpty()) ownedByUser else { + ownedByUser and (SubscriptionGroupMembershipsTable.channelUrl notInList retained) + } + } + } +} diff --git a/src/main/kotlin/dev/typetype/server/services/SubscriptionGroupsService.kt b/src/main/kotlin/dev/typetype/server/services/SubscriptionGroupsService.kt new file mode 100644 index 00000000..23887b2c --- /dev/null +++ b/src/main/kotlin/dev/typetype/server/services/SubscriptionGroupsService.kt @@ -0,0 +1,182 @@ +package dev.typetype.server.services + +import dev.typetype.server.db.DatabaseFactory +import dev.typetype.server.db.tables.SubscriptionGroupMembershipsTable +import dev.typetype.server.db.tables.SubscriptionGroupsTable +import dev.typetype.server.db.tables.SubscriptionsTable +import dev.typetype.server.models.SubscriptionGroupItem +import org.jetbrains.exposed.v1.core.ResultRow +import org.jetbrains.exposed.v1.core.SortOrder +import org.jetbrains.exposed.v1.core.and +import org.jetbrains.exposed.v1.core.eq +import org.jetbrains.exposed.v1.jdbc.deleteWhere +import org.jetbrains.exposed.v1.jdbc.insertIgnore +import org.jetbrains.exposed.v1.jdbc.selectAll +import org.jetbrains.exposed.v1.jdbc.update +import java.sql.SQLException +import java.util.Locale +import java.util.UUID + +class SubscriptionGroupsService { + suspend fun getAll(userId: String): List = DatabaseFactory.query { + val counts = SubscriptionGroupMembershipsTable.selectAll() + .where { SubscriptionGroupMembershipsTable.userId eq userId } + .groupingBy { it[SubscriptionGroupMembershipsTable.groupId] } + .eachCount() + SubscriptionGroupsTable.selectAll() + .where { SubscriptionGroupsTable.userId eq userId } + .orderBy(SubscriptionGroupsTable.createdAt to SortOrder.DESC) + .map { it.toItem(counts[it[SubscriptionGroupsTable.id]] ?: 0) } + } + + suspend fun exists(userId: String, groupId: String): Boolean = DatabaseFactory.query { + groupExists(userId, groupId) + } + + suspend fun create(userId: String, rawName: String): SubscriptionGroupWriteResult { + val name = normalizeDisplayName(rawName) ?: return SubscriptionGroupWriteResult.InvalidName + val normalizedName = normalizeUniqueName(name) + return DatabaseFactory.query { + if (nameExists(userId, normalizedName)) return@query SubscriptionGroupWriteResult.DuplicateName + val id = UUID.randomUUID().toString() + val now = System.currentTimeMillis() + val inserted = SubscriptionGroupsTable.insertIgnore { + it[SubscriptionGroupsTable.id] = id + it[SubscriptionGroupsTable.userId] = userId + it[SubscriptionGroupsTable.name] = name + it[SubscriptionGroupsTable.normalizedName] = normalizedName + it[createdAt] = now + it[updatedAt] = now + }.insertedCount + if (inserted == 0) SubscriptionGroupWriteResult.DuplicateName else { + SubscriptionGroupWriteResult.Success(SubscriptionGroupItem(id, name, 0, now, now)) + } + } + } + + suspend fun rename(userId: String, groupId: String, rawName: String): SubscriptionGroupWriteResult { + val name = normalizeDisplayName(rawName) ?: return SubscriptionGroupWriteResult.InvalidName + val normalizedName = normalizeUniqueName(name) + return try { + DatabaseFactory.query { + val current = SubscriptionGroupsTable.selectAll().where { + (SubscriptionGroupsTable.id eq groupId) and (SubscriptionGroupsTable.userId eq userId) + }.singleOrNull() ?: return@query SubscriptionGroupWriteResult.NotFound + val duplicate = SubscriptionGroupsTable.selectAll().where { + (SubscriptionGroupsTable.userId eq userId) and + (SubscriptionGroupsTable.normalizedName eq normalizedName) + }.any { it[SubscriptionGroupsTable.id] != groupId } + if (duplicate) return@query SubscriptionGroupWriteResult.DuplicateName + val now = System.currentTimeMillis() + SubscriptionGroupsTable.update({ + (SubscriptionGroupsTable.id eq groupId) and (SubscriptionGroupsTable.userId eq userId) + }) { + it[SubscriptionGroupsTable.name] = name + it[SubscriptionGroupsTable.normalizedName] = normalizedName + it[updatedAt] = now + } + val count = membershipCount(userId, groupId) + SubscriptionGroupWriteResult.Success( + SubscriptionGroupItem(groupId, name, count, current[SubscriptionGroupsTable.createdAt], now), + ) + } + } catch (error: Throwable) { + if (error.isUniqueConstraintViolation()) SubscriptionGroupWriteResult.DuplicateName else throw error + } + } + + suspend fun delete(userId: String, groupId: String): Boolean = DatabaseFactory.query { + if (!groupExists(userId, groupId)) return@query false + SubscriptionGroupMembershipsTable.deleteWhere { + (SubscriptionGroupMembershipsTable.groupId eq groupId) and + (SubscriptionGroupMembershipsTable.userId eq userId) + } + SubscriptionGroupsTable.deleteWhere { + (SubscriptionGroupsTable.id eq groupId) and (SubscriptionGroupsTable.userId eq userId) + } > 0 + } + + suspend fun addSubscription( + userId: String, + groupId: String, + rawChannelUrl: String, + ): SubscriptionGroupMembershipResult = DatabaseFactory.query { + if (!groupExists(userId, groupId)) return@query SubscriptionGroupMembershipResult.GroupNotFound + val channelUrl = ChannelUrlCanonicalizer.canonicalize(rawChannelUrl) + val subscriptionExists = SubscriptionsTable.selectAll().where { + (SubscriptionsTable.userId eq userId) and (SubscriptionsTable.channelUrl eq channelUrl) + }.any() + if (!subscriptionExists) return@query SubscriptionGroupMembershipResult.SubscriptionNotFound + SubscriptionGroupMembershipsTable.insertIgnore { + it[SubscriptionGroupMembershipsTable.groupId] = groupId + it[SubscriptionGroupMembershipsTable.userId] = userId + it[SubscriptionGroupMembershipsTable.channelUrl] = channelUrl + it[addedAt] = System.currentTimeMillis() + } + SubscriptionGroupMembershipResult.Success + } + + suspend fun removeSubscription( + userId: String, + groupId: String, + rawChannelUrl: String, + ): SubscriptionGroupMembershipResult = DatabaseFactory.query { + if (!groupExists(userId, groupId)) return@query SubscriptionGroupMembershipResult.GroupNotFound + val channelUrl = ChannelUrlCanonicalizer.canonicalize(rawChannelUrl) + val deleted = SubscriptionGroupMembershipsTable.deleteWhere { + (SubscriptionGroupMembershipsTable.groupId eq groupId) and + (SubscriptionGroupMembershipsTable.userId eq userId) and + (SubscriptionGroupMembershipsTable.channelUrl eq channelUrl) + } + if (deleted > 0) SubscriptionGroupMembershipResult.Success else { + SubscriptionGroupMembershipResult.MembershipNotFound + } + } + + suspend fun getChannelUrls(userId: String, groupId: String): List = DatabaseFactory.query { + SubscriptionGroupMembershipsTable.selectAll().where { + (SubscriptionGroupMembershipsTable.groupId eq groupId) and + (SubscriptionGroupMembershipsTable.userId eq userId) + }.orderBy(SubscriptionGroupMembershipsTable.addedAt to SortOrder.DESC) + .map { it[SubscriptionGroupMembershipsTable.channelUrl] } + } + + private fun groupExists(userId: String, groupId: String): Boolean = + SubscriptionGroupsTable.selectAll().where { + (SubscriptionGroupsTable.id eq groupId) and (SubscriptionGroupsTable.userId eq userId) + }.any() + + private fun nameExists(userId: String, normalizedName: String): Boolean = + SubscriptionGroupsTable.selectAll().where { + (SubscriptionGroupsTable.userId eq userId) and + (SubscriptionGroupsTable.normalizedName eq normalizedName) + }.any() + + private fun membershipCount(userId: String, groupId: String): Int = + SubscriptionGroupMembershipsTable.selectAll().where { + (SubscriptionGroupMembershipsTable.userId eq userId) and + (SubscriptionGroupMembershipsTable.groupId eq groupId) + }.count().toInt() + + private fun ResultRow.toItem(channelCount: Int): SubscriptionGroupItem = SubscriptionGroupItem( + id = this[SubscriptionGroupsTable.id], + name = this[SubscriptionGroupsTable.name], + channelCount = channelCount, + createdAt = this[SubscriptionGroupsTable.createdAt], + updatedAt = this[SubscriptionGroupsTable.updatedAt], + ) + + private fun normalizeDisplayName(value: String): String? = + value.trim().takeIf { it.length in 1..MAX_GROUP_NAME_LENGTH } + + private fun normalizeUniqueName(value: String): String = value.lowercase(Locale.ROOT) + + private fun Throwable.isUniqueConstraintViolation(): Boolean = generateSequence(this) { it.cause } + .filterIsInstance() + .any { it.sqlState == UNIQUE_VIOLATION_SQL_STATE } + + companion object { + const val MAX_GROUP_NAME_LENGTH = 100 + private const val UNIQUE_VIOLATION_SQL_STATE = "23505" + } +} diff --git a/src/main/kotlin/dev/typetype/server/services/SubscriptionSelection.kt b/src/main/kotlin/dev/typetype/server/services/SubscriptionSelection.kt new file mode 100644 index 00000000..cbdf5d99 --- /dev/null +++ b/src/main/kotlin/dev/typetype/server/services/SubscriptionSelection.kt @@ -0,0 +1,17 @@ +package dev.typetype.server.services + +sealed interface SubscriptionSelection { + val cursorKey: String + + data object All : SubscriptionSelection { + override val cursorKey: String = "all" + } + + data object Ungrouped : SubscriptionSelection { + override val cursorKey: String = "ungrouped" + } + + data class Group(val id: String) : SubscriptionSelection { + override val cursorKey: String = "group:$id" + } +} diff --git a/src/main/kotlin/dev/typetype/server/services/SubscriptionsService.kt b/src/main/kotlin/dev/typetype/server/services/SubscriptionsService.kt index 887a931e..21d022de 100644 --- a/src/main/kotlin/dev/typetype/server/services/SubscriptionsService.kt +++ b/src/main/kotlin/dev/typetype/server/services/SubscriptionsService.kt @@ -1,6 +1,7 @@ package dev.typetype.server.services import dev.typetype.server.db.DatabaseFactory +import dev.typetype.server.db.tables.SubscriptionGroupMembershipsTable import dev.typetype.server.db.tables.SubscriptionsTable import dev.typetype.server.models.SubscriptionItem import org.jetbrains.exposed.v1.core.ResultRow @@ -13,14 +14,22 @@ import org.jetbrains.exposed.v1.jdbc.selectAll class SubscriptionsService { - suspend fun getAll(userId: String): List = DatabaseFactory.query { + suspend fun getAll( + userId: String, + selection: SubscriptionSelection = SubscriptionSelection.All, + ): List = DatabaseFactory.query { + val selectedUrls = selectedChannelUrls(userId, selection) val items = SubscriptionsTable.selectAll() .where { SubscriptionsTable.userId eq userId } .orderBy(SubscriptionsTable.subscribedAt to SortOrder.DESC) .map { it.toItem() } + .filter { selection == SubscriptionSelection.All || it.channelUrl in selectedUrls } SubscriptionAvatarRepairer.repair(userId = userId, items = items) } + suspend fun getChannelUrls(userId: String, selection: SubscriptionSelection): Set = + DatabaseFactory.query { selectedChannelUrls(userId, selection) } + suspend fun add(userId: String, item: SubscriptionItem): SubscriptionItem { val canonicalUrl = ChannelUrlCanonicalizer.canonicalize(item.channelUrl) val now = System.currentTimeMillis() @@ -38,9 +47,34 @@ class SubscriptionsService { suspend fun delete(userId: String, channelUrl: String): Boolean = DatabaseFactory.query { val canonicalUrl = ChannelUrlCanonicalizer.canonicalize(channelUrl) + SubscriptionGroupMembershipsTable.deleteWhere { + (SubscriptionGroupMembershipsTable.userId eq userId) and + (SubscriptionGroupMembershipsTable.channelUrl eq canonicalUrl) + } SubscriptionsTable.deleteWhere { SubscriptionsTable.channelUrl eq canonicalUrl and (SubscriptionsTable.userId eq userId) } > 0 } + private fun selectedChannelUrls(userId: String, selection: SubscriptionSelection): Set { + val all = SubscriptionsTable.selectAll() + .where { SubscriptionsTable.userId eq userId } + .mapTo(linkedSetOf()) { ChannelUrlCanonicalizer.canonicalize(it[SubscriptionsTable.channelUrl]) } + if (selection == SubscriptionSelection.All) return all + val memberships = SubscriptionGroupMembershipsTable.selectAll().where { + when (selection) { + SubscriptionSelection.All -> SubscriptionGroupMembershipsTable.userId eq userId + SubscriptionSelection.Ungrouped -> SubscriptionGroupMembershipsTable.userId eq userId + is SubscriptionSelection.Group -> + (SubscriptionGroupMembershipsTable.userId eq userId) and + (SubscriptionGroupMembershipsTable.groupId eq selection.id) + } + }.mapTo(mutableSetOf()) { it[SubscriptionGroupMembershipsTable.channelUrl] } + return when (selection) { + SubscriptionSelection.All -> all + SubscriptionSelection.Ungrouped -> all - memberships + is SubscriptionSelection.Group -> all intersect memberships + } + } + private fun ResultRow.toItem() = SubscriptionItem( channelUrl = ChannelUrlCanonicalizer.canonicalize(this[SubscriptionsTable.channelUrl]), name = this[SubscriptionsTable.name], diff --git a/src/main/kotlin/dev/typetype/server/services/TypeTypeBackupCoreRestore.kt b/src/main/kotlin/dev/typetype/server/services/TypeTypeBackupCoreRestore.kt index f8f5b715..7106218c 100644 --- a/src/main/kotlin/dev/typetype/server/services/TypeTypeBackupCoreRestore.kt +++ b/src/main/kotlin/dev/typetype/server/services/TypeTypeBackupCoreRestore.kt @@ -22,6 +22,7 @@ internal object TypeTypeBackupCoreRestore { this[SubscriptionsTable.avatarUrl] = item.avatarUrl this[SubscriptionsTable.subscribedAt] = item.subscribedAt } + SubscriptionGroupMembershipCleaner.retain(userId, items.map(SubscriptionItem::channelUrl)) return items.size } From 61b01ade3729463f1d266f23e377b090919e40fe Mon Sep 17 00:00:00 2001 From: User Date: Mon, 17 Aug 2026 15:34:38 -0700 Subject: [PATCH 03/14] feat: expose subscription group API Add authenticated CRUD and membership endpoints backed by the account-scoped group service, and register the service with the application. --- .../dev/typetype/server/ServiceRegistry.kt | 2 + .../server/routes/SubscriptionGroupsRoutes.kt | 114 ++++++++++++++++++ 2 files changed, 116 insertions(+) create mode 100644 src/main/kotlin/dev/typetype/server/routes/SubscriptionGroupsRoutes.kt diff --git a/src/main/kotlin/dev/typetype/server/ServiceRegistry.kt b/src/main/kotlin/dev/typetype/server/ServiceRegistry.kt index 3164b3d1..de8e6a0b 100644 --- a/src/main/kotlin/dev/typetype/server/ServiceRegistry.kt +++ b/src/main/kotlin/dev/typetype/server/ServiceRegistry.kt @@ -29,6 +29,7 @@ import dev.typetype.server.services.SubscriptionFeedService import dev.typetype.server.services.SubscriptionShortsBlendService import dev.typetype.server.services.SubscriptionShortsFeedService import dev.typetype.server.services.SubscriptionsService +import dev.typetype.server.services.SubscriptionGroupsService import dev.typetype.server.services.SubscriptionFeedCacheInvalidation import dev.typetype.server.services.SubscriptionFeedCacheInvalidator import dev.typetype.server.services.TypeTypeBackupService @@ -84,6 +85,7 @@ internal class ServiceRegistry( val sabrSessionStore = extraction.sabrSessionStore val historyService = HistoryService() val subscriptionsService = SubscriptionsService() + val subscriptionGroupsService = SubscriptionGroupsService() val subscriptionFeedService = SubscriptionFeedService(subscriptionsService, channelService, cache) val subscriptionShortsFeedService = SubscriptionShortsFeedService( subscriptionsService, diff --git a/src/main/kotlin/dev/typetype/server/routes/SubscriptionGroupsRoutes.kt b/src/main/kotlin/dev/typetype/server/routes/SubscriptionGroupsRoutes.kt new file mode 100644 index 00000000..c80acb3d --- /dev/null +++ b/src/main/kotlin/dev/typetype/server/routes/SubscriptionGroupsRoutes.kt @@ -0,0 +1,114 @@ +package dev.typetype.server.routes + +import dev.typetype.server.models.ErrorResponse +import dev.typetype.server.models.SubscriptionGroupMembershipRequest +import dev.typetype.server.models.SubscriptionGroupRequest +import dev.typetype.server.services.AuthService +import dev.typetype.server.services.SubscriptionGroupMembershipResult +import dev.typetype.server.services.SubscriptionGroupWriteResult +import dev.typetype.server.services.SubscriptionGroupsService +import io.ktor.http.HttpStatusCode +import io.ktor.server.application.ApplicationCall +import io.ktor.server.request.receive +import io.ktor.server.response.respond +import io.ktor.server.routing.Route +import io.ktor.server.routing.delete +import io.ktor.server.routing.get +import io.ktor.server.routing.post +import io.ktor.server.routing.put + +fun Route.subscriptionGroupsRoutes(groupsService: SubscriptionGroupsService, authService: AuthService) { + get("/subscriptions/groups") { + call.withJwtAuth(authService) { userId -> call.respond(groupsService.getAll(userId)) } + } + post("/subscriptions/groups") { + call.withJwtAuth(authService) { userId -> + val request = call.receiveGroupRequest() ?: return@withJwtAuth + call.respondGroupWrite(groupsService.create(userId, request.name), created = true) + } + } + put("/subscriptions/groups/{groupId}") { + call.withJwtAuth(authService) { userId -> + val groupId = call.groupId() ?: return@withJwtAuth call.respondMissingGroupId() + val request = call.receiveGroupRequest() ?: return@withJwtAuth + call.respondGroupWrite(groupsService.rename(userId, groupId, request.name), created = false) + } + } + delete("/subscriptions/groups/{groupId}") { + call.withJwtAuth(authService) { userId -> + val groupId = call.groupId() ?: return@withJwtAuth call.respondMissingGroupId() + if (groupsService.delete(userId, groupId)) call.respond(HttpStatusCode.NoContent) else { + call.respond(HttpStatusCode.NotFound, ErrorResponse("Subscription group not found", "subscription_group_not_found")) + } + } + } + put("/subscriptions/groups/{groupId}/channels") { + call.withJwtAuth(authService) { userId -> + val groupId = call.groupId() ?: return@withJwtAuth call.respondMissingGroupId() + val request = runCatching { call.receive() }.getOrElse { + return@withJwtAuth call.respond(HttpStatusCode.BadRequest, ErrorResponse("Invalid request body")) + } + if (request.channelUrl.isBlank()) { + return@withJwtAuth call.respond(HttpStatusCode.BadRequest, ErrorResponse("channelUrl must not be blank")) + } + call.respondMembership(groupsService.addSubscription(userId, groupId, request.channelUrl)) + } + } + delete("/subscriptions/groups/{groupId}/channels") { + call.withJwtAuth(authService) { userId -> + val groupId = call.groupId() ?: return@withJwtAuth call.respondMissingGroupId() + val channelUrl = call.request.queryParameters["url"]?.takeIf(String::isNotBlank) + ?: return@withJwtAuth call.respond(HttpStatusCode.BadRequest, ErrorResponse("Missing channelUrl")) + call.respondMembership(groupsService.removeSubscription(userId, groupId, channelUrl)) + } + } +} + +private fun ApplicationCall.groupId(): String? = parameters["groupId"]?.takeIf(String::isNotBlank) + +private suspend fun ApplicationCall.receiveGroupRequest(): SubscriptionGroupRequest? = + runCatching { receive() }.getOrElse { + respond(HttpStatusCode.BadRequest, ErrorResponse("Invalid request body")) + null + } + +private suspend fun ApplicationCall.respondGroupWrite(result: SubscriptionGroupWriteResult, created: Boolean) { + when (result) { + is SubscriptionGroupWriteResult.Success -> if (created) respond(HttpStatusCode.Created, result.group) else { + respond(HttpStatusCode.NoContent) + } + SubscriptionGroupWriteResult.InvalidName -> respond( + HttpStatusCode.BadRequest, + ErrorResponse("Group name must contain 1 to 100 characters", "subscription_group_invalid_name"), + ) + SubscriptionGroupWriteResult.DuplicateName -> respond( + HttpStatusCode.Conflict, + ErrorResponse("A subscription group with this name already exists", "subscription_group_name_conflict"), + ) + SubscriptionGroupWriteResult.NotFound -> respond( + HttpStatusCode.NotFound, + ErrorResponse("Subscription group not found", "subscription_group_not_found"), + ) + } +} + +private suspend fun ApplicationCall.respondMembership(result: SubscriptionGroupMembershipResult) { + when (result) { + SubscriptionGroupMembershipResult.Success -> respond(HttpStatusCode.NoContent) + SubscriptionGroupMembershipResult.GroupNotFound -> respond( + HttpStatusCode.NotFound, + ErrorResponse("Subscription group not found", "subscription_group_not_found"), + ) + SubscriptionGroupMembershipResult.SubscriptionNotFound -> respond( + HttpStatusCode.NotFound, + ErrorResponse("Subscription not found", "subscription_not_found"), + ) + SubscriptionGroupMembershipResult.MembershipNotFound -> respond( + HttpStatusCode.NotFound, + ErrorResponse("Subscription group membership not found", "subscription_group_membership_not_found"), + ) + } +} + +private suspend fun ApplicationCall.respondMissingGroupId() = + respond(HttpStatusCode.BadRequest, ErrorResponse("Missing groupId")) From 46126970c810d88ba2d8ad1665d26c4e94ede17c Mon Sep 17 00:00:00 2001 From: User Date: Mon, 17 Aug 2026 15:34:45 -0700 Subject: [PATCH 04/14] feat: filter subscription lists by group Parse group and ungrouped selectors for subscription reads, reject invalid or foreign group IDs, and separate subscription creation input from server-generated timestamps. --- .../models/SubscriptionCreateRequest.kt | 10 ++++++ .../routes/SubscriptionSelectionParameter.kt | 29 +++++++++++++++++ .../server/routes/SubscriptionsRoutes.kt | 32 ++++++++++++++++--- 3 files changed, 67 insertions(+), 4 deletions(-) create mode 100644 src/main/kotlin/dev/typetype/server/models/SubscriptionCreateRequest.kt create mode 100644 src/main/kotlin/dev/typetype/server/routes/SubscriptionSelectionParameter.kt diff --git a/src/main/kotlin/dev/typetype/server/models/SubscriptionCreateRequest.kt b/src/main/kotlin/dev/typetype/server/models/SubscriptionCreateRequest.kt new file mode 100644 index 00000000..0e530fdf --- /dev/null +++ b/src/main/kotlin/dev/typetype/server/models/SubscriptionCreateRequest.kt @@ -0,0 +1,10 @@ +package dev.typetype.server.models + +import kotlinx.serialization.Serializable + +@Serializable +data class SubscriptionCreateRequest( + val channelUrl: String, + val name: String, + val avatarUrl: String, +) diff --git a/src/main/kotlin/dev/typetype/server/routes/SubscriptionSelectionParameter.kt b/src/main/kotlin/dev/typetype/server/routes/SubscriptionSelectionParameter.kt new file mode 100644 index 00000000..a847aab0 --- /dev/null +++ b/src/main/kotlin/dev/typetype/server/routes/SubscriptionSelectionParameter.kt @@ -0,0 +1,29 @@ +package dev.typetype.server.routes + +import dev.typetype.server.services.SubscriptionSelection +import io.ktor.server.application.ApplicationCall + +internal sealed interface SubscriptionSelectionParseResult { + data class Valid(val selection: SubscriptionSelection) : SubscriptionSelectionParseResult + data object Invalid : SubscriptionSelectionParseResult +} + +internal fun ApplicationCall.parseSubscriptionSelection(): SubscriptionSelectionParseResult { + val rawGroupId = request.queryParameters["groupId"] + val groupId = rawGroupId?.takeIf(String::isNotBlank) + if (rawGroupId != null && groupId == null) return SubscriptionSelectionParseResult.Invalid + val rawUngrouped = request.queryParameters["ungrouped"] + val ungrouped = when (rawUngrouped) { + null -> false + "true" -> true + "false" -> false + else -> return SubscriptionSelectionParseResult.Invalid + } + if (groupId != null && ungrouped) return SubscriptionSelectionParseResult.Invalid + val selection = when { + groupId != null -> SubscriptionSelection.Group(groupId) + ungrouped -> SubscriptionSelection.Ungrouped + else -> SubscriptionSelection.All + } + return SubscriptionSelectionParseResult.Valid(selection) +} diff --git a/src/main/kotlin/dev/typetype/server/routes/SubscriptionsRoutes.kt b/src/main/kotlin/dev/typetype/server/routes/SubscriptionsRoutes.kt index 3d7b40b2..becd5d9b 100644 --- a/src/main/kotlin/dev/typetype/server/routes/SubscriptionsRoutes.kt +++ b/src/main/kotlin/dev/typetype/server/routes/SubscriptionsRoutes.kt @@ -1,11 +1,14 @@ package dev.typetype.server.routes import dev.typetype.server.models.ErrorResponse +import dev.typetype.server.models.SubscriptionCreateRequest import dev.typetype.server.models.SubscriptionItem import dev.typetype.server.services.AuthService import dev.typetype.server.services.HomeRecommendationWarmup import dev.typetype.server.services.NoopHomeRecommendationWarmup import dev.typetype.server.services.SubscriptionsService +import dev.typetype.server.services.SubscriptionGroupsService +import dev.typetype.server.services.SubscriptionSelection import io.ktor.http.HttpStatusCode import io.ktor.server.application.ApplicationCall import io.ktor.server.request.receive @@ -18,16 +21,37 @@ import io.ktor.server.routing.post import java.net.URLDecoder import java.nio.charset.StandardCharsets -fun Route.subscriptionsRoutes(subscriptionsService: SubscriptionsService, authService: AuthService, warmupService: HomeRecommendationWarmup = NoopHomeRecommendationWarmup) { +fun Route.subscriptionsRoutes( + subscriptionsService: SubscriptionsService, + authService: AuthService, + warmupService: HomeRecommendationWarmup = NoopHomeRecommendationWarmup, + groupsService: SubscriptionGroupsService = SubscriptionGroupsService(), +) { get("/subscriptions") { - call.withJwtAuth(authService) { userId -> call.respond(subscriptionsService.getAll(userId)) } + call.withJwtAuth(authService) { userId -> + val parsed = call.parseSubscriptionSelection() + if (parsed !is SubscriptionSelectionParseResult.Valid) { + return@withJwtAuth call.respond(HttpStatusCode.BadRequest, ErrorResponse("Invalid subscription filter")) + } + val selection = parsed.selection + if (selection is SubscriptionSelection.Group && !groupsService.exists(userId, selection.id)) { + return@withJwtAuth call.respond( + HttpStatusCode.NotFound, + ErrorResponse("Subscription group not found", "subscription_group_not_found"), + ) + } + call.respond(subscriptionsService.getAll(userId, selection)) + } } post("/subscriptions") { call.withJwtAuth(authService) { userId -> - val item = runCatching { call.receive() }.getOrElse { + val request = runCatching { call.receive() }.getOrElse { return@withJwtAuth call.respond(HttpStatusCode.BadRequest, ErrorResponse("Invalid request body")) } - val subscription = subscriptionsService.add(userId, item) + val subscription = subscriptionsService.add( + userId, + SubscriptionItem(request.channelUrl, request.name, request.avatarUrl), + ) warmupService.invalidateAndWarm(userId) call.respond(HttpStatusCode.Created, subscription) } From 8665282d3d2bc500b987d89609543d71058d5439 Mon Sep 17 00:00:00 2001 From: User Date: Mon, 17 Aug 2026 15:34:53 -0700 Subject: [PATCH 05/14] feat: keep group feed pagination stable Project group and ungrouped feeds from the shared global snapshot while retaining each cursor's account-scoped membership selection in cache for the full pagination session. --- .../server/routes/SubscriptionFeedRoutes.kt | 17 ++++++- .../typetype/server/routes/UserDataRoutes.kt | 15 +++++- .../services/SubscriptionFeedBuilder.kt | 24 +++++++-- .../services/SubscriptionFeedCacheKeys.kt | 2 + .../SubscriptionFeedSelectionStore.kt | 50 +++++++++++++++++++ .../services/SubscriptionFeedService.kt | 19 ++++++- .../services/SubscriptionFeedSnapshot.kt | 50 ++++++++++++++++--- 7 files changed, 163 insertions(+), 14 deletions(-) create mode 100644 src/main/kotlin/dev/typetype/server/services/SubscriptionFeedSelectionStore.kt diff --git a/src/main/kotlin/dev/typetype/server/routes/SubscriptionFeedRoutes.kt b/src/main/kotlin/dev/typetype/server/routes/SubscriptionFeedRoutes.kt index e382da1c..2c3aabc8 100644 --- a/src/main/kotlin/dev/typetype/server/routes/SubscriptionFeedRoutes.kt +++ b/src/main/kotlin/dev/typetype/server/routes/SubscriptionFeedRoutes.kt @@ -6,6 +6,8 @@ import dev.typetype.server.services.AuthService import dev.typetype.server.services.SubscriptionFeedPageResult import dev.typetype.server.services.SubscriptionFeedService import dev.typetype.server.services.SubscriptionFeedVisibility +import dev.typetype.server.services.SubscriptionGroupsService +import dev.typetype.server.services.SubscriptionSelection import dev.typetype.server.services.SettingsService import io.ktor.http.HttpHeaders import io.ktor.http.HttpStatusCode @@ -19,12 +21,24 @@ fun Route.subscriptionFeedRoutes( feedService: SubscriptionFeedService, authService: AuthService, settingsService: SettingsService? = null, + groupsService: SubscriptionGroupsService = SubscriptionGroupsService(), ) { get("/subscriptions/feed") { call.withJwtAuth(authService) { userId -> + val parsed = call.parseSubscriptionSelection() + if (parsed !is SubscriptionSelectionParseResult.Valid) { + return@withJwtAuth call.respond(HttpStatusCode.BadRequest, ErrorResponse("Invalid subscription filter")) + } + val selection = parsed.selection + val cursor = call.request.queryParameters["cursor"] + if (cursor == null && selection is SubscriptionSelection.Group && !groupsService.exists(userId, selection.id)) { + return@withJwtAuth call.respond( + HttpStatusCode.NotFound, + ErrorResponse("Subscription group not found", "subscription_group_not_found"), + ) + } val page = call.request.queryParameters["page"]?.toIntOrNull()?.coerceIn(0, MAX_FEED_PAGE) ?: 0 val limit = call.request.queryParameters["limit"]?.toIntOrNull()?.coerceIn(1, 100) ?: 30 - val cursor = call.request.queryParameters["cursor"] val visibility = settingsService?.subscriptionFeedVisibility(userId) ?: SubscriptionFeedVisibility() call.response.headers.append(HttpHeaders.CacheControl, "no-store") when ( @@ -35,6 +49,7 @@ fun Route.subscriptionFeedRoutes( cursor, visibility.hideLiveStreams, visibility.hideMembersOnlyContent, + selection, ) ) { is SubscriptionFeedPageResult.Ready -> call.respond(result.response) diff --git a/src/main/kotlin/dev/typetype/server/routes/UserDataRoutes.kt b/src/main/kotlin/dev/typetype/server/routes/UserDataRoutes.kt index 74baab91..1ff721c8 100644 --- a/src/main/kotlin/dev/typetype/server/routes/UserDataRoutes.kt +++ b/src/main/kotlin/dev/typetype/server/routes/UserDataRoutes.kt @@ -17,8 +17,19 @@ internal fun Route.userDataRoutes( restoreService: PipePipeBackupImporterService, ) { historyRoutes(svc.historyService, authService, svc.settingsService) - subscriptionsRoutes(svc.subscriptionsService, authService, svc.homeRecommendationWarmupService) - subscriptionFeedRoutes(svc.subscriptionFeedService, authService, svc.settingsService) + subscriptionGroupsRoutes(svc.subscriptionGroupsService, authService) + subscriptionsRoutes( + svc.subscriptionsService, + authService, + svc.homeRecommendationWarmupService, + svc.subscriptionGroupsService, + ) + subscriptionFeedRoutes( + svc.subscriptionFeedService, + authService, + svc.settingsService, + svc.subscriptionGroupsService, + ) subscriptionShortsFeedRoutes(svc.subscriptionShortsFeedService, authService) rssFeedRoutes(svc.rssFeedManagementService, authService) playlistRoutes(svc.playlistService, authService, svc.videoMetadataRepairService) diff --git a/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedBuilder.kt b/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedBuilder.kt index bae327dc..91621e5d 100644 --- a/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedBuilder.kt +++ b/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedBuilder.kt @@ -22,13 +22,28 @@ internal class SubscriptionFeedBuilder(private val channelService: ChannelServic } catch (error: CancellationException) { throw error } catch (_: Throwable) { - SubscriptionSourceResult(emptyList(), successfulSources = 0, failedSources = 1) + SubscriptionSourceResult( + channelUrl = subscription.channelUrl, + videos = emptyList(), + successfulSources = 0, + failedSources = 1, + ) } } }.map { it.await() } - val videos = outcomes.flatMap { it.videos }.deduplicated() + val videosByKey = linkedMapOf() + val sourceChannelUrls = linkedMapOf>() + outcomes.forEach { outcome -> + outcome.videos.forEach { video -> + val key = video.subscriptionFeedKey() + val current = videosByKey[key] + if (current == null || video.isLive && !current.isLive) videosByKey[key] = video + sourceChannelUrls.getOrPut(key, ::linkedSetOf).add(outcome.channelUrl) + } + } SubscriptionFeedBuildResult( - videos = videos, + videos = videosByKey.values.toList(), + sourceChannelUrls = sourceChannelUrls.mapValues { it.value.toList() }, successfulSources = outcomes.sumOf { it.successfulSources }, failedSources = outcomes.sumOf { it.failedSources }, ) @@ -46,6 +61,7 @@ internal class SubscriptionFeedBuilder(private val channelService: ChannelServic } val results = listOfNotNull(channelResult, liveResult) SubscriptionSourceResult( + channelUrl = channelUrl, videos = mergeVideos(videos), successfulSources = results.count { it.success }, failedSources = results.count { !it.success }, @@ -91,6 +107,7 @@ internal class SubscriptionFeedBuilder(private val channelService: ChannelServic private data class SourceFetchResult(val videos: List, val success: Boolean) private data class SubscriptionSourceResult( + val channelUrl: String, val videos: List, val successfulSources: Int, val failedSources: Int, @@ -105,6 +122,7 @@ internal class SubscriptionFeedBuilder(private val channelService: ChannelServic internal data class SubscriptionFeedBuildResult( val videos: List, + val sourceChannelUrls: Map>, val successfulSources: Int, val failedSources: Int, ) diff --git a/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedCacheKeys.kt b/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedCacheKeys.kt index d305a5da..af5f1eb0 100644 --- a/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedCacheKeys.kt +++ b/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedCacheKeys.kt @@ -9,6 +9,8 @@ object SubscriptionFeedCacheKeys { fun invalidation(userId: String): String = "feed:invalidation:${hash(userId)}" + fun selection(userId: String, token: String): String = "feed:selection:${hash(userId)}:$token" + fun shorts(userId: String): String = "feed:shorts:${hash(userId)}" private fun hash(userId: String): String = MessageDigest diff --git a/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedSelectionStore.kt b/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedSelectionStore.kt new file mode 100644 index 00000000..27d42252 --- /dev/null +++ b/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedSelectionStore.kt @@ -0,0 +1,50 @@ +package dev.typetype.server.services + +import dev.typetype.server.cache.CacheJson +import dev.typetype.server.cache.CacheService +import kotlinx.serialization.Serializable +import java.util.UUID + +internal class SubscriptionFeedSelectionStore( + private val cache: CacheService, + private val subscriptions: SubscriptionsService, +) { + suspend fun resolve( + userId: String, + selection: SubscriptionSelection, + token: String?, + ): SubscriptionFeedSelectionSnapshot? { + if (selection == SubscriptionSelection.All) return SubscriptionFeedSelectionSnapshot(null, null) + if (token == null) { + val channelUrls = subscriptions.getChannelUrls(userId, selection) + val nextToken = UUID.randomUUID().toString() + cache.set( + SubscriptionFeedCacheKeys.selection(userId, nextToken), + CacheJson.encodeToString( + StoredSubscriptionFeedSelection.serializer(), + StoredSubscriptionFeedSelection(selection.cursorKey, channelUrls.toList()), + ), + SubscriptionFeedSnapshotStore.RETENTION_SECONDS, + ) + return SubscriptionFeedSelectionSnapshot(nextToken, channelUrls) + } + val raw = runCatching { cache.get(SubscriptionFeedCacheKeys.selection(userId, token)) }.getOrNull() + ?: return null + val stored = runCatching { + CacheJson.decodeFromString(StoredSubscriptionFeedSelection.serializer(), raw) + }.getOrNull() ?: return null + if (stored.filterKey != selection.cursorKey) return null + return SubscriptionFeedSelectionSnapshot(token, stored.channelUrls.toSet()) + } +} + +@Serializable +private data class StoredSubscriptionFeedSelection( + val filterKey: String, + val channelUrls: List, +) + +internal data class SubscriptionFeedSelectionSnapshot( + val token: String?, + val channelUrls: Set?, +) diff --git a/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedService.kt b/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedService.kt index de1da91c..38e75d8d 100644 --- a/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedService.kt +++ b/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedService.kt @@ -26,6 +26,7 @@ class SubscriptionFeedService( private val refreshScope: CoroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.IO), ) { private val store = SubscriptionFeedSnapshotStore(cache, clock) + private val selections = SubscriptionFeedSelectionStore(cache, subscriptionsService) private val builder = SubscriptionFeedBuilder(channelService) private val orderer = SubscriptionFeedOrderer() private val refreshJobs = ConcurrentHashMap() @@ -38,6 +39,7 @@ class SubscriptionFeedService( cursor: String?, hideLiveStreams: Boolean = false, hideMembersOnlyContent: Boolean = false, + selection: SubscriptionSelection = SubscriptionSelection.All, requestId: String? = currentRequestId(), ): SubscriptionFeedPageResult { val current = store.current(userId) @@ -57,6 +59,9 @@ class SubscriptionFeedService( if (cursorState != null && cursorState.hideMembersOnlyContent != hideMembersOnlyContent) { return SubscriptionFeedPageResult.InvalidCursor } + if (cursorState != null && cursorState.filterKey != selection.cursorKey) { + return SubscriptionFeedPageResult.InvalidCursor + } val snapshot = when { cursorState == null -> current cursorState.generation == current.generation -> current @@ -64,8 +69,19 @@ class SubscriptionFeedService( ?: return SubscriptionFeedPageResult.StaleGeneration } val offset = cursorState?.offset ?: page * limit + val selected = selections.resolve(userId, selection, cursorState?.selectionToken) + ?: return SubscriptionFeedPageResult.StaleGeneration return SubscriptionFeedPageResult.Ready( - snapshot.page(offset, limit, isRefreshing(userId), hideLiveStreams, hideMembersOnlyContent), + snapshot.page( + offset, + limit, + isRefreshing(userId), + hideLiveStreams, + hideMembersOnlyContent, + selection, + selected.channelUrls, + selected.token, + ), ) } @@ -152,6 +168,7 @@ class SubscriptionFeedService( stale = false, videos = ordering.videos, livePromotedAt = ordering.livePromotedAt, + sourceChannelUrls = result.sourceChannelUrls, ) runCatching { store.publish(userId, snapshot) }.onFailure { logger.warn("subscription_feed event=publish_failed user={} error={}", userKey(userId), it.message) diff --git a/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedSnapshot.kt b/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedSnapshot.kt index c752034a..5d8232a1 100644 --- a/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedSnapshot.kt +++ b/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedSnapshot.kt @@ -13,6 +13,7 @@ internal data class SubscriptionFeedSnapshot( val stale: Boolean, val videos: List, val livePromotedAt: Map = emptyMap(), + val sourceChannelUrls: Map> = emptyMap(), ) @Serializable @@ -22,6 +23,8 @@ private data class SubscriptionFeedCursor( val limit: Int, val hideLiveStreams: Boolean = false, val hideMembersOnlyContent: Boolean = false, + val filterKey: String = SubscriptionSelection.All.cursorKey, + val selectionToken: String? = null, ) internal object SubscriptionFeedCursorCodec { @@ -31,10 +34,14 @@ internal object SubscriptionFeedCursorCodec { limit: Int, hideLiveStreams: Boolean, hideMembersOnlyContent: Boolean, + filterKey: String, + selectionToken: String?, ): String { val payload = CacheJson.encodeToString( SubscriptionFeedCursor.serializer(), - SubscriptionFeedCursor(generation, offset, limit, hideLiveStreams, hideMembersOnlyContent), + SubscriptionFeedCursor( + generation, offset, limit, hideLiveStreams, hideMembersOnlyContent, filterKey, selectionToken, + ), ) return Base64.getUrlEncoder().withoutPadding().encodeToString(payload.toByteArray()) } @@ -42,7 +49,10 @@ internal object SubscriptionFeedCursorCodec { fun decode(value: String): SubscriptionFeedCursorState? = runCatching { val payload = String(Base64.getUrlDecoder().decode(value)) val cursor = CacheJson.decodeFromString(SubscriptionFeedCursor.serializer(), payload) - cursor.takeIf { it.generation > 0L && it.offset >= 0 && it.limit in 1..100 } + cursor.takeIf { + it.generation > 0L && it.offset >= 0 && it.limit in 1..100 && + ((it.filterKey == SubscriptionSelection.All.cursorKey) == (it.selectionToken == null)) + } ?.let { SubscriptionFeedCursorState( it.generation, @@ -50,6 +60,8 @@ internal object SubscriptionFeedCursorCodec { it.limit, it.hideLiveStreams, it.hideMembersOnlyContent, + it.filterKey, + it.selectionToken, ) } }.getOrNull() @@ -61,6 +73,8 @@ internal data class SubscriptionFeedCursorState( val limit: Int, val hideLiveStreams: Boolean, val hideMembersOnlyContent: Boolean, + val filterKey: String, + val selectionToken: String?, ) internal fun SubscriptionFeedSnapshot.page( @@ -69,26 +83,31 @@ internal fun SubscriptionFeedSnapshot.page( refreshing: Boolean, hideLiveStreams: Boolean = false, hideMembersOnlyContent: Boolean = false, + selection: SubscriptionSelection = SubscriptionSelection.All, + selectedChannelUrls: Set? = null, + selectionToken: String? = null, ): SubscriptionFeedResponse { - val visibleVideos = videos.filterNot { video -> + val projectedVideos = projectedVideos(selection, selectedChannelUrls).filterNot { video -> (hideLiveStreams && video.isLiveOrUpcomingAt(generatedAt)) || (hideMembersOnlyContent && video.requiresMembership) } - val from = offset.coerceAtMost(visibleVideos.size) - val to = minOf(from + limit, visibleVideos.size) - val nextpage = if (to < visibleVideos.size) { + val from = offset.coerceAtMost(projectedVideos.size) + val to = minOf(from + limit, projectedVideos.size) + val nextpage = if (to < projectedVideos.size) { SubscriptionFeedCursorCodec.encode( generation, to, limit, hideLiveStreams, hideMembersOnlyContent, + selection.cursorKey, + selectionToken, ) } else { null } return SubscriptionFeedResponse( - videos = visibleVideos.subList(from, to), + videos = projectedVideos.subList(from, to), nextpage = nextpage, generation = generation, generatedAt = generatedAt, @@ -96,6 +115,23 @@ internal fun SubscriptionFeedSnapshot.page( ) } +private fun SubscriptionFeedSnapshot.projectedVideos( + selection: SubscriptionSelection, + selectedChannelUrls: Set?, +): List { + if (selection == SubscriptionSelection.All) return videos + val allowed = selectedChannelUrls.orEmpty() + if (allowed.isEmpty()) return emptyList() + return videos.filter { video -> + val sources = sourceChannelUrls[video.subscriptionFeedKey()] + if (sources != null) { + sources.any { ChannelUrlCanonicalizer.canonicalize(it) in allowed } + } else { + ChannelUrlCanonicalizer.canonicalize(video.uploaderUrl) in allowed + } + } +} + internal sealed interface SubscriptionFeedPageResult { data class Ready(val response: SubscriptionFeedResponse) : SubscriptionFeedPageResult data class Preparing(val retryAfterMs: Long) : SubscriptionFeedPageResult From 860d6869709165b98bf4a12a616fd68e85e7c763 Mon Sep 17 00:00:00 2001 From: User Date: Mon, 17 Aug 2026 15:35:00 -0700 Subject: [PATCH 06/14] feat: preserve subscription groups in backups Export named groups with their channel memberships and restore them transactionally with subscriptions. Validate names and membership references before replacing account-owned group data. --- openapi/components/user-backup.yaml | 12 ++++ .../models/SubscriptionGroupBackupItem.kt | 11 ++++ .../server/models/TypeTypeBackupItem.kt | 1 + .../SubscriptionGroupBackupRepository.kt | 64 +++++++++++++++++++ .../services/TypeTypeBackupRestoreWriter.kt | 5 ++ .../server/services/TypeTypeBackupService.kt | 37 +++++++++++ .../server/TypeTypeBackupServiceTest.kt | 19 ++++++ 7 files changed, 149 insertions(+) create mode 100644 src/main/kotlin/dev/typetype/server/models/SubscriptionGroupBackupItem.kt create mode 100644 src/main/kotlin/dev/typetype/server/services/SubscriptionGroupBackupRepository.kt diff --git a/openapi/components/user-backup.yaml b/openapi/components/user-backup.yaml index 36f5a1d4..76428dd8 100644 --- a/openapi/components/user-backup.yaml +++ b/openapi/components/user-backup.yaml @@ -23,6 +23,17 @@ TypeTypeContentFiltersBackup: allowedPlaylists: type: array items: { $ref: ./access-control.yaml#/AllowedPlaylistItem } +SubscriptionGroupBackupItem: + type: object + required: [name, channelUrls, createdAt, updatedAt] + properties: + name: { type: string, minLength: 1, maxLength: 100 } + channelUrls: + type: array + uniqueItems: true + items: { type: string, minLength: 1 } + createdAt: { type: integer, format: int64 } + updatedAt: { type: integer, format: int64 } TypeTypeBackupItem: type: object required: [format, version, exportedAt, categories] @@ -47,6 +58,7 @@ TypeTypeBackupItem: - settings - contentFilters subscriptions: { type: array, nullable: true, items: { type: object, additionalProperties: true } } + subscriptionGroups: { type: array, nullable: true, items: { $ref: '#/SubscriptionGroupBackupItem' } } history: { type: array, nullable: true, items: { type: object, additionalProperties: true } } playlists: { type: array, nullable: true, items: { type: object, additionalProperties: true } } watchLater: { type: array, nullable: true, items: { type: object, additionalProperties: true } } diff --git a/src/main/kotlin/dev/typetype/server/models/SubscriptionGroupBackupItem.kt b/src/main/kotlin/dev/typetype/server/models/SubscriptionGroupBackupItem.kt new file mode 100644 index 00000000..efae0cab --- /dev/null +++ b/src/main/kotlin/dev/typetype/server/models/SubscriptionGroupBackupItem.kt @@ -0,0 +1,11 @@ +package dev.typetype.server.models + +import kotlinx.serialization.Serializable + +@Serializable +data class SubscriptionGroupBackupItem( + val name: String, + val channelUrls: List, + val createdAt: Long, + val updatedAt: Long, +) diff --git a/src/main/kotlin/dev/typetype/server/models/TypeTypeBackupItem.kt b/src/main/kotlin/dev/typetype/server/models/TypeTypeBackupItem.kt index 4c9300b9..82cb9f40 100644 --- a/src/main/kotlin/dev/typetype/server/models/TypeTypeBackupItem.kt +++ b/src/main/kotlin/dev/typetype/server/models/TypeTypeBackupItem.kt @@ -9,6 +9,7 @@ data class TypeTypeBackupItem( val exportedAt: Long, val categories: List, val subscriptions: List? = null, + val subscriptionGroups: List? = null, val history: List? = null, val playlists: List? = null, val watchLater: List? = null, diff --git a/src/main/kotlin/dev/typetype/server/services/SubscriptionGroupBackupRepository.kt b/src/main/kotlin/dev/typetype/server/services/SubscriptionGroupBackupRepository.kt new file mode 100644 index 00000000..3cc685aa --- /dev/null +++ b/src/main/kotlin/dev/typetype/server/services/SubscriptionGroupBackupRepository.kt @@ -0,0 +1,64 @@ +package dev.typetype.server.services + +import dev.typetype.server.db.DatabaseFactory +import dev.typetype.server.db.tables.SubscriptionGroupMembershipsTable +import dev.typetype.server.db.tables.SubscriptionGroupsTable +import dev.typetype.server.models.SubscriptionGroupBackupItem +import org.jetbrains.exposed.v1.core.SortOrder +import org.jetbrains.exposed.v1.core.eq +import org.jetbrains.exposed.v1.jdbc.batchInsert +import org.jetbrains.exposed.v1.jdbc.deleteWhere +import org.jetbrains.exposed.v1.jdbc.selectAll +import java.util.Locale +import java.util.UUID + +internal object SubscriptionGroupBackupRepository { + suspend fun export(userId: String): List = DatabaseFactory.query { + val channelsByGroup = SubscriptionGroupMembershipsTable.selectAll() + .where { SubscriptionGroupMembershipsTable.userId eq userId } + .orderBy(SubscriptionGroupMembershipsTable.addedAt to SortOrder.ASC) + .groupBy( + keySelector = { it[SubscriptionGroupMembershipsTable.groupId] }, + valueTransform = { it[SubscriptionGroupMembershipsTable.channelUrl] }, + ) + SubscriptionGroupsTable.selectAll() + .where { SubscriptionGroupsTable.userId eq userId } + .orderBy(SubscriptionGroupsTable.createdAt to SortOrder.ASC) + .map { row -> + SubscriptionGroupBackupItem( + name = row[SubscriptionGroupsTable.name], + channelUrls = channelsByGroup[row[SubscriptionGroupsTable.id]].orEmpty(), + createdAt = row[SubscriptionGroupsTable.createdAt], + updatedAt = row[SubscriptionGroupsTable.updatedAt], + ) + } + } + + fun restore(userId: String, items: List): Pair { + SubscriptionGroupMembershipsTable.deleteWhere { SubscriptionGroupMembershipsTable.userId eq userId } + SubscriptionGroupsTable.deleteWhere { SubscriptionGroupsTable.userId eq userId } + val groups = items.map { it to UUID.randomUUID().toString() } + if (groups.isNotEmpty()) { + SubscriptionGroupsTable.batchInsert(groups, shouldReturnGeneratedValues = false) { (item, id) -> + this[SubscriptionGroupsTable.id] = id + this[SubscriptionGroupsTable.userId] = userId + this[SubscriptionGroupsTable.name] = item.name + this[SubscriptionGroupsTable.normalizedName] = item.name.lowercase(Locale.ROOT) + this[SubscriptionGroupsTable.createdAt] = item.createdAt + this[SubscriptionGroupsTable.updatedAt] = item.updatedAt + } + } + val memberships = groups.flatMap { (item, groupId) -> + item.channelUrls.map { channelUrl -> groupId to ChannelUrlCanonicalizer.canonicalize(channelUrl) } + } + if (memberships.isNotEmpty()) { + SubscriptionGroupMembershipsTable.batchInsert(memberships, shouldReturnGeneratedValues = false) { (groupId, channelUrl) -> + this[SubscriptionGroupMembershipsTable.groupId] = groupId + this[SubscriptionGroupMembershipsTable.userId] = userId + this[SubscriptionGroupMembershipsTable.channelUrl] = channelUrl + this[SubscriptionGroupMembershipsTable.addedAt] = System.currentTimeMillis() + } + } + return groups.size to memberships.size + } +} diff --git a/src/main/kotlin/dev/typetype/server/services/TypeTypeBackupRestoreWriter.kt b/src/main/kotlin/dev/typetype/server/services/TypeTypeBackupRestoreWriter.kt index ad10644a..d951dafd 100644 --- a/src/main/kotlin/dev/typetype/server/services/TypeTypeBackupRestoreWriter.kt +++ b/src/main/kotlin/dev/typetype/server/services/TypeTypeBackupRestoreWriter.kt @@ -16,6 +16,11 @@ internal object TypeTypeBackupRestoreWriter { userId, requireNotNull(backup.subscriptions), ) + backup.subscriptionGroups?.let { groups -> + val counts = SubscriptionGroupBackupRepository.restore(userId, groups) + restored["subscriptionGroups"] = counts.first + restored["subscriptionGroupMemberships"] = counts.second + } } if (TypeTypeBackupCategory.HISTORY in categories) { restored["history"] = TypeTypeBackupCoreRestore.history( diff --git a/src/main/kotlin/dev/typetype/server/services/TypeTypeBackupService.kt b/src/main/kotlin/dev/typetype/server/services/TypeTypeBackupService.kt index 1c98ddfb..97466980 100644 --- a/src/main/kotlin/dev/typetype/server/services/TypeTypeBackupService.kt +++ b/src/main/kotlin/dev/typetype/server/services/TypeTypeBackupService.kt @@ -5,6 +5,7 @@ import dev.typetype.server.models.TYPE_TYPE_BACKUP_VERSION import dev.typetype.server.models.TypeTypeBackupItem import dev.typetype.server.models.TypeTypeContentFiltersBackup import dev.typetype.server.models.TypeTypeRestoreSummary +import java.util.Locale class TypeTypeBackupService( private val subscriptions: SubscriptionsService, @@ -34,6 +35,11 @@ class TypeTypeBackupService( exportedAt = System.currentTimeMillis(), categories = categories.map(TypeTypeBackupCategory::wireName).sorted(), subscriptions = if (includes(TypeTypeBackupCategory.SUBSCRIPTIONS)) subscriptions.getAll(userId) else null, + subscriptionGroups = if (includes(TypeTypeBackupCategory.SUBSCRIPTIONS)) { + SubscriptionGroupBackupRepository.export(userId) + } else { + null + }, history = if (includes(TypeTypeBackupCategory.HISTORY)) history.getAll(userId) else null, playlists = fullPlaylists, watchLater = if (includes(TypeTypeBackupCategory.WATCH_LATER)) watchLater.getAll(userId) else null, @@ -53,6 +59,7 @@ class TypeTypeBackupService( val categories = TypeTypeBackupCategory.parse(backup.categories.joinToString(",")) ?: throw IllegalArgumentException("Invalid backup categories") validateSections(backup, categories) + validateSubscriptionGroups(backup, categories) validateContentFilters(backup, categories) return TypeTypeBackupRestoreWriter.restore(userId, backup, categories) } @@ -87,6 +94,36 @@ private fun validateSections( require(missing.isEmpty()) { "Backup is missing selected data" } } +private fun validateSubscriptionGroups( + backup: TypeTypeBackupItem, + categories: Set, +) { + val groups = backup.subscriptionGroups ?: return + require(TypeTypeBackupCategory.SUBSCRIPTIONS in categories) { + "Subscription groups require the subscriptions category" + } + val normalizedNames = groups.map { group -> + require(group.name == group.name.trim() && group.name.length in 1..SubscriptionGroupsService.MAX_GROUP_NAME_LENGTH) { + "Subscription group names must contain 1 to 100 characters" + } + group.name.lowercase(Locale.ROOT) + } + require(normalizedNames.distinct().size == normalizedNames.size) { + "Backup contains duplicate subscription group names" + } + val subscriptions = requireNotNull(backup.subscriptions) + .mapTo(mutableSetOf()) { ChannelUrlCanonicalizer.canonicalize(it.channelUrl) } + groups.forEach { group -> + val channels = group.channelUrls.map(ChannelUrlCanonicalizer::canonicalize) + require(channels.distinct().size == channels.size) { + "Backup contains duplicate subscription group memberships" + } + require(channels.all { it in subscriptions }) { + "Subscription group membership references an unknown subscription" + } + } +} + private fun validateContentFilters( backup: TypeTypeBackupItem, categories: Set, diff --git a/src/test/kotlin/dev/typetype/server/TypeTypeBackupServiceTest.kt b/src/test/kotlin/dev/typetype/server/TypeTypeBackupServiceTest.kt index 8cce3567..7d9a2219 100644 --- a/src/test/kotlin/dev/typetype/server/TypeTypeBackupServiceTest.kt +++ b/src/test/kotlin/dev/typetype/server/TypeTypeBackupServiceTest.kt @@ -23,6 +23,9 @@ import dev.typetype.server.services.ProgressService import dev.typetype.server.services.SavedPlaylistService import dev.typetype.server.services.SearchHistoryService import dev.typetype.server.services.SettingsService +import dev.typetype.server.services.SubscriptionGroupMembershipResult +import dev.typetype.server.services.SubscriptionGroupsService +import dev.typetype.server.services.SubscriptionGroupWriteResult import dev.typetype.server.services.SubscriptionsService import dev.typetype.server.services.TypeTypeBackupCategory import dev.typetype.server.services.TypeTypeBackupService @@ -37,6 +40,7 @@ import org.junit.jupiter.api.Test class TypeTypeBackupServiceTest { private val subscriptions = SubscriptionsService() + private val subscriptionGroups = SubscriptionGroupsService() private val history = HistoryService() private val playlists = PlaylistService() private val watchLater = WatchLaterService() @@ -78,6 +82,13 @@ class TypeTypeBackupServiceTest { @Test fun `full backup restores every user data category`() = runTest { subscriptions.add(SOURCE, SubscriptionItem("https://youtube.com/channel/source", "Source", "avatar")) + val subscriptionGroup = ( + subscriptionGroups.create(SOURCE, "Favorites") as SubscriptionGroupWriteResult.Success + ).group + assertEquals( + SubscriptionGroupMembershipResult.Success, + subscriptionGroups.addSubscription(SOURCE, subscriptionGroup.id, "https://youtube.com/channel/source"), + ) history.addImported(SOURCE, videoHistory()) val playlist = playlists.create(SOURCE, PlaylistItem(name = "Saved videos")) playlists.addVideo(SOURCE, playlist.id, playlistVideo()) @@ -107,6 +118,14 @@ class TypeTypeBackupServiceTest { val result = service.restore(TARGET, backup) assertEquals(1, result.restored["subscriptions"]) + assertEquals(1, result.restored["subscriptionGroups"]) + assertEquals(1, result.restored["subscriptionGroupMemberships"]) + val restoredGroup = subscriptionGroups.getAll(TARGET).single() + assertEquals("Favorites", restoredGroup.name) + assertEquals( + listOf("https://youtube.com/channel/source"), + subscriptionGroups.getChannelUrls(TARGET, restoredGroup.id), + ) assertEquals(1, result.restored["history"]) assertEquals(1, result.restored["playlists"]) assertEquals(1, result.restored["playlistVideos"]) From f756bfab3dbf9e5eaeca2399fa4282ae8138c897 Mon Sep 17 00:00:00 2001 From: User Date: Mon, 17 Aug 2026 15:35:05 -0700 Subject: [PATCH 07/14] docs: document subscription group endpoints Describe group management and filtered list/feed operations, and make subscription creation use a request schema without the server-generated subscribedAt field. --- openapi.yaml | 8 + openapi/components/subscriptions.yaml | 34 ++++ openapi/paths/subscriptions.yaml | 157 +++++++++++++++++- .../server/SubscriptionsRoutesTest.kt | 16 ++ 4 files changed, 214 insertions(+), 1 deletion(-) diff --git a/openapi.yaml b/openapi.yaml index f551cc64..37be9bbe 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -43,6 +43,10 @@ paths: /playlist: { $ref: ./openapi/paths/playlists.yaml#/Playlist } /saved-playlists: { $ref: ./openapi/paths/saved-playlists.yaml#/SavedPlaylists } /saved-playlists/{id}: { $ref: ./openapi/paths/saved-playlists.yaml#/SavedPlaylist } + /subscriptions: { $ref: ./openapi/paths/subscriptions.yaml#/Subscriptions } + /subscriptions/groups: { $ref: ./openapi/paths/subscriptions.yaml#/SubscriptionGroups } + /subscriptions/groups/{groupId}: { $ref: ./openapi/paths/subscriptions.yaml#/SubscriptionGroup } + /subscriptions/groups/{groupId}/channels: { $ref: ./openapi/paths/subscriptions.yaml#/SubscriptionGroupChannels } /subscriptions/feed: { $ref: ./openapi/paths/subscriptions.yaml#/SubscriptionFeed } /rss/feeds: { $ref: ./openapi/paths/rss.yaml#/RssFeeds } /rss/feeds/{id}: { $ref: ./openapi/paths/rss.yaml#/RssFeed } @@ -146,6 +150,10 @@ components: $ref: ./openapi/components/media.yaml#/PublicPlaylistItem SavedPlaylistItem: { $ref: ./openapi/components/media.yaml#/SavedPlaylistItem } SavedPlaylistRequest: { $ref: ./openapi/components/media.yaml#/SavedPlaylistRequest } + SubscriptionItem: { $ref: ./openapi/components/subscriptions.yaml#/SubscriptionItem } + SubscriptionGroupItem: { $ref: ./openapi/components/subscriptions.yaml#/SubscriptionGroupItem } + SubscriptionGroupRequest: { $ref: ./openapi/components/subscriptions.yaml#/SubscriptionGroupRequest } + SubscriptionGroupMembershipRequest: { $ref: ./openapi/components/subscriptions.yaml#/SubscriptionGroupMembershipRequest } SubscriptionFeedResponse: { $ref: ./openapi/components/subscriptions.yaml#/SubscriptionFeedResponse } SubscriptionFeedPreparingResponse: { $ref: ./openapi/components/subscriptions.yaml#/SubscriptionFeedPreparingResponse } RssFeedRequest: { $ref: ./openapi/components/rss.yaml#/RssFeedRequest } diff --git a/openapi/components/subscriptions.yaml b/openapi/components/subscriptions.yaml index b0e7ad59..5894cc5f 100644 --- a/openapi/components/subscriptions.yaml +++ b/openapi/components/subscriptions.yaml @@ -1,3 +1,37 @@ +SubscriptionItem: + type: object + required: [channelUrl, name, avatarUrl, subscribedAt] + properties: + channelUrl: { type: string, minLength: 1 } + name: { type: string } + avatarUrl: { type: string } + subscribedAt: { type: integer, format: int64 } +SubscriptionCreateRequest: + type: object + required: [channelUrl, name, avatarUrl] + properties: + channelUrl: { type: string, minLength: 1 } + name: { type: string } + avatarUrl: { type: string } +SubscriptionGroupItem: + type: object + required: [id, name, channelCount, createdAt, updatedAt] + properties: + id: { type: string, format: uuid } + name: { type: string, minLength: 1, maxLength: 100 } + channelCount: { type: integer, minimum: 0 } + createdAt: { type: integer, format: int64 } + updatedAt: { type: integer, format: int64 } +SubscriptionGroupRequest: + type: object + required: [name] + properties: + name: { type: string, minLength: 1, maxLength: 100 } +SubscriptionGroupMembershipRequest: + type: object + required: [channelUrl] + properties: + channelUrl: { type: string, minLength: 1 } SubscriptionFeedResponse: type: object required: [videos, nextpage, generation, generatedAt, refreshing] diff --git a/openapi/paths/subscriptions.yaml b/openapi/paths/subscriptions.yaml index f0dcc65c..ab9bee92 100644 --- a/openapi/paths/subscriptions.yaml +++ b/openapi/paths/subscriptions.yaml @@ -1,8 +1,161 @@ +Subscriptions: + get: + tags: [user-data] + summary: List the current user's subscriptions + description: Omit both filters for the global list. Use groupId for one named group or ungrouped=true for subscriptions in no groups. + parameters: + - name: groupId + in: query + required: false + schema: { type: string, format: uuid } + - name: ungrouped + in: query + required: false + schema: { type: boolean, default: false } + responses: + '200': + description: The selected subscription projection. + content: + application/json: + schema: + type: array + items: { $ref: ../components/subscriptions.yaml#/SubscriptionItem } + '400': { $ref: ../components/common.yaml#/JsonError } + '401': { $ref: ../components/common.yaml#/JsonError } + '404': { $ref: ../components/common.yaml#/JsonError } + post: + tags: [user-data] + summary: Subscribe to a channel + requestBody: + required: true + content: + application/json: + schema: { $ref: ../components/subscriptions.yaml#/SubscriptionCreateRequest } + responses: + '201': + description: Subscription created. + content: + application/json: + schema: { $ref: ../components/subscriptions.yaml#/SubscriptionItem } + '400': { $ref: ../components/common.yaml#/JsonError } + '401': { $ref: ../components/common.yaml#/JsonError } + delete: + tags: [user-data] + summary: Unsubscribe from a channel + parameters: + - name: url + in: query + required: true + schema: { type: string, minLength: 1 } + responses: + '204': { description: Subscription deleted. } + '400': { $ref: ../components/common.yaml#/JsonError } + '401': { $ref: ../components/common.yaml#/JsonError } + '404': { $ref: ../components/common.yaml#/JsonError } +SubscriptionGroups: + get: + tags: [user-data] + summary: List the current user's subscription groups + responses: + '200': + description: Account-scoped named groups. + content: + application/json: + schema: + type: array + items: { $ref: ../components/subscriptions.yaml#/SubscriptionGroupItem } + '401': { $ref: ../components/common.yaml#/JsonError } + post: + tags: [user-data] + summary: Create a subscription group + requestBody: + required: true + content: + application/json: + schema: { $ref: ../components/subscriptions.yaml#/SubscriptionGroupRequest } + responses: + '201': + description: Group created. + content: + application/json: + schema: { $ref: ../components/subscriptions.yaml#/SubscriptionGroupItem } + '400': { $ref: ../components/common.yaml#/JsonError } + '401': { $ref: ../components/common.yaml#/JsonError } + '409': { $ref: ../components/common.yaml#/JsonError } +SubscriptionGroup: + parameters: + - name: groupId + in: path + required: true + schema: { type: string, format: uuid } + put: + tags: [user-data] + summary: Rename a subscription group + requestBody: + required: true + content: + application/json: + schema: { $ref: ../components/subscriptions.yaml#/SubscriptionGroupRequest } + responses: + '204': { description: Group renamed. } + '400': { $ref: ../components/common.yaml#/JsonError } + '401': { $ref: ../components/common.yaml#/JsonError } + '404': { $ref: ../components/common.yaml#/JsonError } + '409': { $ref: ../components/common.yaml#/JsonError } + delete: + tags: [user-data] + summary: Delete a subscription group + responses: + '204': { description: Group and its memberships deleted. } + '401': { $ref: ../components/common.yaml#/JsonError } + '404': { $ref: ../components/common.yaml#/JsonError } +SubscriptionGroupChannels: + parameters: + - name: groupId + in: path + required: true + schema: { type: string, format: uuid } + put: + tags: [user-data] + summary: Add a subscribed channel to a group + requestBody: + required: true + content: + application/json: + schema: { $ref: ../components/subscriptions.yaml#/SubscriptionGroupMembershipRequest } + responses: + '204': { description: Membership exists. } + '400': { $ref: ../components/common.yaml#/JsonError } + '401': { $ref: ../components/common.yaml#/JsonError } + '404': { $ref: ../components/common.yaml#/JsonError } + delete: + tags: [user-data] + summary: Remove a subscribed channel from a group + parameters: + - name: url + in: query + required: true + schema: { type: string, minLength: 1 } + responses: + '204': { description: Membership deleted. } + '400': { $ref: ../components/common.yaml#/JsonError } + '401': { $ref: ../components/common.yaml#/JsonError } + '404': { $ref: ../components/common.yaml#/JsonError } SubscriptionFeed: get: tags: [user-data] summary: Read a stable page from the current user's subscription feed snapshot parameters: + - name: groupId + in: query + required: false + description: Restrict the snapshot projection to subscriptions in one account-owned group. + schema: { type: string, format: uuid } + - name: ungrouped + in: query + required: false + description: Restrict the snapshot projection to subscriptions in no groups. + schema: { type: boolean, default: false } - name: page in: query required: false @@ -16,7 +169,7 @@ SubscriptionFeed: - name: cursor in: query required: false - description: Opaque continuation returned in nextpage. + description: Opaque continuation returned in nextpage and bound to the selected membership snapshot. schema: { type: string } responses: '200': @@ -45,6 +198,8 @@ SubscriptionFeed: $ref: ../components/common.yaml#/JsonError '401': $ref: ../components/common.yaml#/JsonError + '404': + $ref: ../components/common.yaml#/JsonError '409': description: The cursor references a snapshot generation that is no longer retained. headers: diff --git a/src/test/kotlin/dev/typetype/server/SubscriptionsRoutesTest.kt b/src/test/kotlin/dev/typetype/server/SubscriptionsRoutesTest.kt index d3075356..df2da9ac 100644 --- a/src/test/kotlin/dev/typetype/server/SubscriptionsRoutesTest.kt +++ b/src/test/kotlin/dev/typetype/server/SubscriptionsRoutesTest.kt @@ -20,10 +20,13 @@ import io.ktor.server.routing.routing import io.ktor.server.testing.ApplicationTestBuilder import io.ktor.server.testing.testApplication import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse import org.junit.jupiter.api.Assertions.assertTrue import org.junit.jupiter.api.BeforeAll import org.junit.jupiter.api.BeforeEach import org.junit.jupiter.api.Test +import java.nio.file.Files +import java.nio.file.Path class SubscriptionsRoutesTest { @@ -49,6 +52,19 @@ class SubscriptionsRoutesTest { private val itemBody = """{"channelUrl":"https://yt.com/channel/1","name":"Test","avatarUrl":""}""" + @Test + fun `subscription creation contract omits the server timestamp`() { + val components = Files.readString(Path.of("openapi/components/subscriptions.yaml")) + val requestSchema = components + .substringAfter("SubscriptionCreateRequest:") + .substringBefore("SubscriptionGroupItem:") + val paths = Files.readString(Path.of("openapi/paths/subscriptions.yaml")) + + assertTrue("required: [channelUrl, name, avatarUrl]" in requestSchema) + assertFalse("subscribedAt" in requestSchema) + assertTrue("#/SubscriptionCreateRequest" in paths) + } + @Test fun `GET subscriptions without token returns 401`() = withApp { assertEquals(HttpStatusCode.Unauthorized, client.get("/subscriptions").status) From 8336b00362762dab731ba64327a01b3931078a01 Mon Sep 17 00:00:00 2001 From: User Date: Mon, 17 Aug 2026 15:35:11 -0700 Subject: [PATCH 08/14] test: cover subscription group persistence Verify account isolation, normalized unique names, many-to-many membership, ungrouped selection, and cleanup after subscription replacement or deletion. --- .../server/SubscriptionGroupsServiceTest.kt | 120 ++++++++++++++++++ 1 file changed, 120 insertions(+) create mode 100644 src/test/kotlin/dev/typetype/server/SubscriptionGroupsServiceTest.kt diff --git a/src/test/kotlin/dev/typetype/server/SubscriptionGroupsServiceTest.kt b/src/test/kotlin/dev/typetype/server/SubscriptionGroupsServiceTest.kt new file mode 100644 index 00000000..d03bce44 --- /dev/null +++ b/src/test/kotlin/dev/typetype/server/SubscriptionGroupsServiceTest.kt @@ -0,0 +1,120 @@ +package dev.typetype.server + +import dev.typetype.server.models.SubscriptionItem +import dev.typetype.server.db.DatabaseFactory +import dev.typetype.server.services.SubscriptionGroupMembershipCleaner +import dev.typetype.server.services.SubscriptionGroupMembershipResult +import dev.typetype.server.services.SubscriptionGroupWriteResult +import dev.typetype.server.services.SubscriptionGroupsService +import dev.typetype.server.services.SubscriptionSelection +import dev.typetype.server.services.SubscriptionsService +import kotlinx.coroutines.test.runTest +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.BeforeAll +import org.junit.jupiter.api.BeforeEach +import org.junit.jupiter.api.Test + +class SubscriptionGroupsServiceTest { + private val groups = SubscriptionGroupsService() + private val subscriptions = SubscriptionsService() + + companion object { + @BeforeAll + @JvmStatic + fun initDb() = TestDatabase.setup() + } + + @BeforeEach + fun clean() = TestDatabase.truncateAll() + + @Test + fun `group names are normalized unique and account scoped`() = runTest { + val group = groups.create("user-a", " Work ").createdGroup() + + assertEquals("Work", group.name) + assertEquals(SubscriptionGroupWriteResult.DuplicateName, groups.create("user-a", "work")) + assertTrue(groups.create("user-b", "work") is SubscriptionGroupWriteResult.Success) + assertFalse(groups.exists("user-b", group.id)) + assertEquals( + SubscriptionGroupWriteResult.NotFound, + groups.rename("user-b", group.id, "Other"), + ) + groups.create("user-a", "Other") + assertEquals(SubscriptionGroupWriteResult.DuplicateName, groups.rename("user-a", group.id, "OTHER")) + } + + @Test + fun `a subscription can belong to multiple groups while ungrouped stays distinct`() = runTest { + subscriptions.add("user", subscription("one")) + subscriptions.add("user", subscription("two")) + subscriptions.add("user", subscription("three")) + val first = groups.create("user", "First").createdGroup() + val second = groups.create("user", "Second").createdGroup() + + assertEquals(SubscriptionGroupMembershipResult.Success, groups.addSubscription("user", first.id, channel("one"))) + assertEquals(SubscriptionGroupMembershipResult.Success, groups.addSubscription("user", second.id, channel("one"))) + assertEquals(SubscriptionGroupMembershipResult.Success, groups.addSubscription("user", second.id, channel("two"))) + assertEquals(1, groups.getAll("user").first { it.id == first.id }.channelCount) + + assertEquals( + listOf(channel("one")), + subscriptions.getAll("user", SubscriptionSelection.Group(first.id)).map { it.channelUrl }, + ) + assertEquals( + setOf(channel("one"), channel("two")), + subscriptions.getAll("user", SubscriptionSelection.Group(second.id)).map { it.channelUrl }.toSet(), + ) + assertEquals( + listOf(channel("three")), + subscriptions.getAll("user", SubscriptionSelection.Ungrouped).map { it.channelUrl }, + ) + } + + @Test + fun `membership requires both the users group and subscription`() = runTest { + val group = groups.create("user-a", "A").createdGroup() + subscriptions.add("user-b", subscription("shared")) + + assertEquals( + SubscriptionGroupMembershipResult.SubscriptionNotFound, + groups.addSubscription("user-a", group.id, channel("shared")), + ) + assertEquals( + SubscriptionGroupMembershipResult.GroupNotFound, + groups.addSubscription("user-b", group.id, channel("shared")), + ) + } + + @Test + fun `deleting a subscription removes its memberships`() = runTest { + val group = groups.create("user", "Group").createdGroup() + subscriptions.add("user", subscription("one")) + groups.addSubscription("user", group.id, channel("one")) + + assertTrue(subscriptions.delete("user", channel("one"))) + + assertEquals(emptyList(), groups.getChannelUrls("user", group.id)) + } + + @Test + fun `replacement imports retain only memberships for subscriptions still present`() = runTest { + val group = groups.create("user", "Group").createdGroup() + subscriptions.add("user", subscription("one")) + subscriptions.add("user", subscription("two")) + groups.addSubscription("user", group.id, channel("one")) + groups.addSubscription("user", group.id, channel("two")) + + DatabaseFactory.query { SubscriptionGroupMembershipCleaner.retain("user", listOf(channel("one"))) } + + assertEquals(listOf(channel("one")), groups.getChannelUrls("user", group.id)) + } + + private fun SubscriptionGroupWriteResult.createdGroup() = + (this as SubscriptionGroupWriteResult.Success).group + + private fun subscription(id: String) = SubscriptionItem(channel(id), id, "") + + private fun channel(id: String) = "https://yt.com/channel/$id" +} From 939158ab4a2e43aabec5dc266f10ec4b6d260026 Mon Sep 17 00:00:00 2001 From: User Date: Mon, 17 Aug 2026 15:35:16 -0700 Subject: [PATCH 09/14] test: cover subscription group routes Exercise authenticated group CRUD, membership updates, filter validation, and cross-account access through the HTTP routing surface. --- .../server/SubscriptionGroupsRoutesTest.kt | 161 ++++++++++++++++++ 1 file changed, 161 insertions(+) create mode 100644 src/test/kotlin/dev/typetype/server/SubscriptionGroupsRoutesTest.kt diff --git a/src/test/kotlin/dev/typetype/server/SubscriptionGroupsRoutesTest.kt b/src/test/kotlin/dev/typetype/server/SubscriptionGroupsRoutesTest.kt new file mode 100644 index 00000000..70dd04b3 --- /dev/null +++ b/src/test/kotlin/dev/typetype/server/SubscriptionGroupsRoutesTest.kt @@ -0,0 +1,161 @@ +package dev.typetype.server + +import dev.typetype.server.models.SubscriptionGroupItem +import dev.typetype.server.models.SubscriptionItem +import dev.typetype.server.routes.subscriptionGroupsRoutes +import dev.typetype.server.routes.subscriptionsRoutes +import dev.typetype.server.services.AuthService +import dev.typetype.server.services.SubscriptionGroupsService +import dev.typetype.server.services.SubscriptionsService +import io.ktor.client.request.delete +import io.ktor.client.request.get +import io.ktor.client.request.header +import io.ktor.client.request.parameter +import io.ktor.client.request.post +import io.ktor.client.request.put +import io.ktor.client.request.setBody +import io.ktor.client.statement.bodyAsText +import io.ktor.http.ContentType +import io.ktor.http.HttpHeaders +import io.ktor.http.HttpStatusCode +import io.ktor.serialization.kotlinx.json.json +import io.ktor.server.application.install +import io.ktor.server.plugins.contentnegotiation.ContentNegotiation +import io.ktor.server.routing.routing +import io.ktor.server.testing.ApplicationTestBuilder +import io.ktor.server.testing.testApplication +import kotlinx.serialization.json.Json +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.BeforeAll +import org.junit.jupiter.api.BeforeEach +import org.junit.jupiter.api.Test + +class SubscriptionGroupsRoutesTest { + private val groups = SubscriptionGroupsService() + private val subscriptions = SubscriptionsService() + private val auth = AuthService.fixed(TEST_USER_ID) + + companion object { + @BeforeAll + @JvmStatic + fun initDb() = TestDatabase.setup() + } + + @BeforeEach + fun clean() = TestDatabase.truncateAll() + + private fun withApp(block: suspend ApplicationTestBuilder.() -> Unit) = testApplication { + application { + install(ContentNegotiation) { json() } + routing { + subscriptionGroupsRoutes(groups, auth) + subscriptionsRoutes(subscriptions, auth, groupsService = groups) + } + } + block() + } + + @Test + fun `group routes require authentication`() = withApp { + assertEquals(HttpStatusCode.Unauthorized, client.get("/subscriptions/groups").status) + } + + @Test + fun `groups can be created listed renamed and deleted`() = withApp { + val create = client.post("/subscriptions/groups") { + authorizeJson() + setBody("""{"name":"Work"}""") + } + assertEquals(HttpStatusCode.Created, create.status) + val group = Json.decodeFromString(create.bodyAsText()) + + assertTrue(authorizedGet("/subscriptions/groups").bodyAsText().contains("\"name\":\"Work\"")) + assertEquals(HttpStatusCode.NoContent, client.put("/subscriptions/groups/${group.id}") { + authorizeJson() + setBody("""{"name":"Research"}""") + }.status) + assertTrue(authorizedGet("/subscriptions/groups").bodyAsText().contains("\"name\":\"Research\"")) + assertEquals(HttpStatusCode.NoContent, client.delete("/subscriptions/groups/${group.id}") { authorize() }.status) + assertEquals("[]", authorizedGet("/subscriptions/groups").bodyAsText()) + } + + @Test + fun `blank and duplicate group names are rejected`() = withApp { + assertEquals(HttpStatusCode.BadRequest, client.post("/subscriptions/groups") { + authorizeJson() + setBody("""{"name":" "}""") + }.status) + assertEquals(HttpStatusCode.Created, client.post("/subscriptions/groups") { + authorizeJson() + setBody("""{"name":"Work"}""") + }.status) + assertEquals(HttpStatusCode.Conflict, client.post("/subscriptions/groups") { + authorizeJson() + setBody("""{"name":"work"}""") + }.status) + } + + @Test + fun `membership drives grouped and ungrouped subscription projections`() = withApp { + subscriptions.add(TEST_USER_ID, SubscriptionItem(channel("one"), "One", "")) + subscriptions.add(TEST_USER_ID, SubscriptionItem(channel("two"), "Two", "")) + val group = createGroup("Work") + + assertEquals(HttpStatusCode.NoContent, client.put("/subscriptions/groups/${group.id}/channels") { + authorizeJson() + setBody("""{"channelUrl":"${channel("one")}"}""") + }.status) + + val grouped = authorizedGet("/subscriptions") { parameter("groupId", group.id) } + assertTrue(grouped.bodyAsText().contains(channel("one"))) + assertTrue(!grouped.bodyAsText().contains(channel("two"))) + val ungrouped = authorizedGet("/subscriptions") { parameter("ungrouped", true) } + assertTrue(!ungrouped.bodyAsText().contains(channel("one"))) + assertTrue(ungrouped.bodyAsText().contains(channel("two"))) + + assertEquals(HttpStatusCode.NoContent, client.delete("/subscriptions/groups/${group.id}/channels") { + authorize() + parameter("url", channel("one")) + }.status) + assertTrue(authorizedGet("/subscriptions") { parameter("ungrouped", true) }.bodyAsText().contains(channel("one"))) + } + + @Test + fun `invalid or inaccessible filters fail explicitly`() = withApp { + assertEquals(HttpStatusCode.BadRequest, authorizedGet("/subscriptions") { + parameter("groupId", "group") + parameter("ungrouped", true) + }.status) + assertEquals(HttpStatusCode.NotFound, authorizedGet("/subscriptions") { + parameter("groupId", "missing") + }.status) + } + + private suspend fun ApplicationTestBuilder.createGroup(name: String): SubscriptionGroupItem { + val response = client.post("/subscriptions/groups") { + authorizeJson() + setBody("""{"name":"$name"}""") + } + return Json.decodeFromString(response.bodyAsText()) + } + + private suspend fun ApplicationTestBuilder.authorizedGet( + path: String, + configure: io.ktor.client.request.HttpRequestBuilder.() -> Unit = {}, + ) = client.get(path) { + authorize() + configure() + } + + private fun io.ktor.client.request.HttpRequestBuilder.authorize() { + header(HttpHeaders.Authorization, "Bearer test-jwt") + } + + private fun io.ktor.client.request.HttpRequestBuilder.authorizeJson() { + authorize() + header(HttpHeaders.ContentType, ContentType.Application.Json.toString()) + } + + private fun channel(id: String) = "https://yt.com/channel/$id" +} From af4cd5c1e607ceef051f66f80db59c3562ad8a4c Mon Sep 17 00:00:00 2001 From: User Date: Mon, 17 Aug 2026 15:35:21 -0700 Subject: [PATCH 10/14] test: cover stable subscription group feeds Verify shared-snapshot projection, source-channel attribution, filter-bound cursors, and unchanged membership snapshots across paginated group reads. --- .../server/SubscriptionGroupFeedRoutesTest.kt | 168 ++++++++++++++++++ 1 file changed, 168 insertions(+) create mode 100644 src/test/kotlin/dev/typetype/server/SubscriptionGroupFeedRoutesTest.kt diff --git a/src/test/kotlin/dev/typetype/server/SubscriptionGroupFeedRoutesTest.kt b/src/test/kotlin/dev/typetype/server/SubscriptionGroupFeedRoutesTest.kt new file mode 100644 index 00000000..c2becbaa --- /dev/null +++ b/src/test/kotlin/dev/typetype/server/SubscriptionGroupFeedRoutesTest.kt @@ -0,0 +1,168 @@ +package dev.typetype.server + +import dev.typetype.server.models.SubscriptionFeedResponse +import dev.typetype.server.models.SubscriptionItem +import dev.typetype.server.routes.subscriptionFeedRoutes +import dev.typetype.server.services.AuthService +import dev.typetype.server.services.SubscriptionFeedService +import dev.typetype.server.services.SubscriptionGroupMembershipResult +import dev.typetype.server.services.SubscriptionGroupWriteResult +import dev.typetype.server.services.SubscriptionGroupsService +import dev.typetype.server.services.SubscriptionsService +import io.ktor.client.request.get +import io.ktor.client.request.header +import io.ktor.client.request.parameter +import io.ktor.client.statement.HttpResponse +import io.ktor.client.statement.bodyAsText +import io.ktor.http.HttpHeaders +import io.ktor.http.HttpStatusCode +import io.ktor.serialization.kotlinx.json.json +import io.ktor.server.application.install +import io.ktor.server.plugins.contentnegotiation.ContentNegotiation +import io.ktor.server.routing.routing +import io.ktor.server.testing.ApplicationTestBuilder +import io.ktor.server.testing.testApplication +import io.mockk.coEvery +import io.mockk.mockk +import kotlinx.serialization.json.Json +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.BeforeAll +import org.junit.jupiter.api.BeforeEach +import org.junit.jupiter.api.Test + +class SubscriptionGroupFeedRoutesTest { + private val subscriptions = SubscriptionsService() + private val groups = SubscriptionGroupsService() + private lateinit var feed: SubscriptionFeedService + private val auth = AuthService.fixed(TEST_USER_ID) + + companion object { + @BeforeAll + @JvmStatic + fun initDb() = TestDatabase.setup() + } + + @BeforeEach + fun clean() { + TestDatabase.truncateAll() + feed = SubscriptionFeedService(subscriptions, FakeChannelService(), FakeCacheService()) + } + + private fun withApp(block: suspend ApplicationTestBuilder.() -> Unit) = testApplication { + application { + install(ContentNegotiation) { json() } + routing { subscriptionFeedRoutes(feed, auth, groupsService = groups) } + } + block() + } + + @Test + fun `group and ungrouped feeds project one shared global snapshot`() = withApp { + subscriptions.add(TEST_USER_ID, subscription("one")) + subscriptions.add(TEST_USER_ID, subscription("two")) + val group = (groups.create(TEST_USER_ID, "Work") as SubscriptionGroupWriteResult.Success).group + assertEquals( + SubscriptionGroupMembershipResult.Success, + groups.addSubscription(TEST_USER_ID, group.id, channel("one")), + ) + + assertEquals(HttpStatusCode.Accepted, requestFeed(groupId = group.id).status) + feed.awaitRefresh(TEST_USER_ID) + + assertEquals(listOf("${channel("one")}/video"), requestReadyFeed(groupId = group.id).videos.map { it.url }) + assertEquals(listOf("${channel("two")}/video"), requestReadyFeed(ungrouped = true).videos.map { it.url }) + assertEquals(2, requestReadyFeed().videos.size) + } + + @Test + fun `cursor keeps its original group membership across pages`() = withApp { + val channelService = mockk() + coEvery { channelService.getChannel(channel("one"), null) } returns SubscriptionFeedTestFixtures.channel( + SubscriptionFeedTestFixtures.video(3_000L, channel = "one", url = "video-one"), + ) + coEvery { channelService.getChannel(channel("two"), null) } returns SubscriptionFeedTestFixtures.channel( + SubscriptionFeedTestFixtures.video(2_000L, channel = "two", url = "video-two"), + ) + coEvery { channelService.getChannel(channel("three"), null) } returns SubscriptionFeedTestFixtures.channel( + SubscriptionFeedTestFixtures.video(1_000L, channel = "three", url = "video-three"), + ) + feed = SubscriptionFeedService(subscriptions, channelService, FakeCacheService()) + listOf("one", "two", "three").forEach { subscriptions.add(TEST_USER_ID, subscription(it)) } + val group = (groups.create(TEST_USER_ID, "Work") as SubscriptionGroupWriteResult.Success).group + groups.addSubscription(TEST_USER_ID, group.id, channel("one")) + groups.addSubscription(TEST_USER_ID, group.id, channel("two")) + assertEquals(HttpStatusCode.Accepted, requestFeed(limit = 1, groupId = group.id).status) + feed.awaitRefresh(TEST_USER_ID) + val firstPage = requestReadyFeed(limit = 1, groupId = group.id) + assertEquals(listOf("video-one"), firstPage.videos.map { it.url }) + + groups.removeSubscription(TEST_USER_ID, group.id, channel("two")) + groups.addSubscription(TEST_USER_ID, group.id, channel("three")) + val secondPage = requestFeed(limit = 1, cursor = requireNotNull(firstPage.nextpage), groupId = group.id) + + assertEquals(HttpStatusCode.OK, secondPage.status) + assertEquals(listOf("video-two"), Json.decodeFromString(secondPage.bodyAsText()).videos.map { it.url }) + } + + @Test + fun `cursor cannot be reused with another subscription filter`() = withApp { + subscriptions.add(TEST_USER_ID, subscription("one")) + subscriptions.add(TEST_USER_ID, subscription("two")) + val group = (groups.create(TEST_USER_ID, "Work") as SubscriptionGroupWriteResult.Success).group + groups.addSubscription(TEST_USER_ID, group.id, channel("one")) + assertEquals(HttpStatusCode.Accepted, requestFeed(limit = 1).status) + feed.awaitRefresh(TEST_USER_ID) + val cursor = requireNotNull(requestReadyFeed(limit = 1).nextpage) + + val response = requestFeed(limit = 1, cursor = cursor, groupId = group.id) + + assertEquals(HttpStatusCode.BadRequest, response.status) + assertTrue(response.bodyAsText().contains("subscription_feed_invalid_cursor")) + } + + @Test + fun `group feed follows the fetched subscription source when uploader url differs`() = withApp { + val sourceUrl = channel("one") + subscriptions.add(TEST_USER_ID, subscription("one")) + val group = (groups.create(TEST_USER_ID, "Work") as SubscriptionGroupWriteResult.Success).group + groups.addSubscription(TEST_USER_ID, group.id, sourceUrl) + val channelService = mockk() + coEvery { channelService.getChannel(sourceUrl, null) } returns SubscriptionFeedTestFixtures.channel( + SubscriptionFeedTestFixtures.video(1_000L, channel = "different-canonical-uploader"), + ) + feed = SubscriptionFeedService(subscriptions, channelService, FakeCacheService()) + + assertEquals(HttpStatusCode.Accepted, requestFeed(groupId = group.id).status) + feed.awaitRefresh(TEST_USER_ID) + + assertEquals(1, requestReadyFeed(groupId = group.id).videos.size) + } + + private suspend fun ApplicationTestBuilder.requestReadyFeed( + limit: Int = 30, + groupId: String? = null, + ungrouped: Boolean = false, + ): SubscriptionFeedResponse { + val response = requestFeed(limit = limit, groupId = groupId, ungrouped = ungrouped) + assertEquals(HttpStatusCode.OK, response.status) + return Json.decodeFromString(response.bodyAsText()) + } + + private suspend fun ApplicationTestBuilder.requestFeed( + limit: Int = 30, + cursor: String? = null, + groupId: String? = null, + ungrouped: Boolean = false, + ): HttpResponse = client.get("/subscriptions/feed") { + header(HttpHeaders.Authorization, "Bearer test-jwt") + parameter("limit", limit) + cursor?.let { parameter("cursor", it) } + groupId?.let { parameter("groupId", it) } + if (ungrouped) parameter("ungrouped", true) + } + + private fun subscription(id: String) = SubscriptionItem(channel(id), id, "") + + private fun channel(id: String) = "https://example.com/channel/$id" +} From abd1ae9aec6e19120ad4bd5fb66e3c1c5c46e183 Mon Sep 17 00:00:00 2001 From: User Date: Mon, 17 Aug 2026 16:09:13 -0700 Subject: [PATCH 11/14] fix: bound feed membership snapshot storage Filtered pagination needs a stable membership view without creating an unbounded cache entry for every initial request. Allocate a fixed set of account-scoped slots atomically, reuse content-derived tokens, and reject a new distinct session at capacity without evicting any issued cursor. Strengthen observable compatibility coverage for legacy backups and server timestamps. Constraint: Every issued cursor must retain its membership snapshot for the full cache TTL Rejected: Evict the oldest snapshot after eight sessions | invalidates a still-live cursor Rejected: Store all snapshots in one read-modify-write value | loses concurrent writes across server instances Rejected: Put channel URLs directly in the cursor | produces oversized client-controlled cursor payloads Confidence: high Scope-risk: moderate Reversibility: clean Directive: Never overwrite an occupied selection slot; reject new sessions before weakening issued cursors Tested: Focused feed, concurrent selection-store, subscription route, backup, and OpenAPI tests on JDK 25 Not-tested: Live Dragonfly slot saturation before the final runtime gate --- openapi/paths/subscriptions.yaml | 9 ++ .../dev/typetype/server/cache/CacheService.kt | 4 + .../typetype/server/cache/DragonflyService.kt | 19 +++++ .../server/routes/SubscriptionFeedRoutes.kt | 8 ++ .../services/SubscriptionFeedCacheKeys.kt | 2 +- .../SubscriptionFeedSelectionStore.kt | 84 +++++++++++++++---- .../services/SubscriptionFeedService.kt | 24 +++--- .../services/SubscriptionFeedSnapshot.kt | 6 +- .../dev/typetype/server/FakeCacheService.kt | 14 ++++ .../SubscriptionFeedSelectionStoreTest.kt | 73 ++++++++++++++++ .../server/SubscriptionGroupFeedRoutesTest.kt | 26 +++++- .../server/SubscriptionsRoutesTest.kt | 40 ++++----- .../server/TypeTypeBackupServiceTest.kt | 25 ++++++ 13 files changed, 285 insertions(+), 49 deletions(-) create mode 100644 src/test/kotlin/dev/typetype/server/SubscriptionFeedSelectionStoreTest.kt diff --git a/openapi/paths/subscriptions.yaml b/openapi/paths/subscriptions.yaml index ab9bee92..9f9b4863 100644 --- a/openapi/paths/subscriptions.yaml +++ b/openapi/paths/subscriptions.yaml @@ -209,3 +209,12 @@ SubscriptionFeed: application/json: schema: $ref: ../components/common.yaml#/ErrorResponse + '429': + description: The account already has the maximum number of active filtered cursor sessions. + headers: + X-Request-ID: + $ref: ../components/common.yaml#/RequestIdHeader + content: + application/json: + schema: + $ref: ../components/common.yaml#/ErrorResponse diff --git a/src/main/kotlin/dev/typetype/server/cache/CacheService.kt b/src/main/kotlin/dev/typetype/server/cache/CacheService.kt index cd67eb79..40f16770 100644 --- a/src/main/kotlin/dev/typetype/server/cache/CacheService.kt +++ b/src/main/kotlin/dev/typetype/server/cache/CacheService.kt @@ -3,5 +3,9 @@ package dev.typetype.server.cache interface CacheService { suspend fun get(key: String): String? suspend fun set(key: String, value: String, ttlSeconds: Long) + suspend fun setIfAbsent(key: String, value: String, ttlSeconds: Long): Boolean = + throw UnsupportedOperationException("Atomic set-if-absent is not supported") + suspend fun refreshIfValueMatches(key: String, value: String, ttlSeconds: Long): Boolean = + throw UnsupportedOperationException("Atomic compare-and-expire is not supported") suspend fun delete(key: String) } diff --git a/src/main/kotlin/dev/typetype/server/cache/DragonflyService.kt b/src/main/kotlin/dev/typetype/server/cache/DragonflyService.kt index 32aa51b2..d8517ccd 100644 --- a/src/main/kotlin/dev/typetype/server/cache/DragonflyService.kt +++ b/src/main/kotlin/dev/typetype/server/cache/DragonflyService.kt @@ -1,6 +1,8 @@ package dev.typetype.server.cache import io.lettuce.core.RedisClient +import io.lettuce.core.ScriptOutputType +import io.lettuce.core.SetArgs import io.lettuce.core.api.StatefulRedisConnection import io.lettuce.core.api.async.RedisAsyncCommands import kotlinx.coroutines.future.await @@ -17,8 +19,25 @@ class DragonflyService(url: String) : CacheService { override suspend fun set(key: String, value: String, ttlSeconds: Long): Unit = async.setex(key, ttlSeconds, value).await().let {} + override suspend fun setIfAbsent(key: String, value: String, ttlSeconds: Long): Boolean = + async.set(key, value, SetArgs.Builder.nx().ex(ttlSeconds)).await() == "OK" + + override suspend fun refreshIfValueMatches(key: String, value: String, ttlSeconds: Long): Boolean = + async.eval( + REFRESH_IF_VALUE_MATCHES, + ScriptOutputType.INTEGER, + arrayOf(key), + value, + ttlSeconds.toString(), + ).await() == 1L + override suspend fun delete(key: String): Unit = async.del(key).await().let {} suspend fun ping(): Boolean = async.ping().await() == "PONG" + + private companion object { + const val REFRESH_IF_VALUE_MATCHES = + "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('expire', KEYS[1], ARGV[2]) end return 0" + } } diff --git a/src/main/kotlin/dev/typetype/server/routes/SubscriptionFeedRoutes.kt b/src/main/kotlin/dev/typetype/server/routes/SubscriptionFeedRoutes.kt index 2c3aabc8..abdfb136 100644 --- a/src/main/kotlin/dev/typetype/server/routes/SubscriptionFeedRoutes.kt +++ b/src/main/kotlin/dev/typetype/server/routes/SubscriptionFeedRoutes.kt @@ -2,6 +2,7 @@ package dev.typetype.server.routes import dev.typetype.server.models.ErrorResponse import dev.typetype.server.models.SubscriptionFeedPreparingResponse +import dev.typetype.server.preserveTooManyRequestsBody import dev.typetype.server.services.AuthService import dev.typetype.server.services.SubscriptionFeedPageResult import dev.typetype.server.services.SubscriptionFeedService @@ -68,6 +69,13 @@ fun Route.subscriptionFeedRoutes( HttpStatusCode.Conflict, ErrorResponse("Subscription feed generation is no longer available", "subscription_feed_stale_generation"), ) + SubscriptionFeedPageResult.CursorCapacityReached -> { + call.preserveTooManyRequestsBody() + call.respond( + HttpStatusCode.TooManyRequests, + ErrorResponse("Too many active subscription feed cursors", "subscription_feed_cursor_capacity"), + ) + } } } } diff --git a/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedCacheKeys.kt b/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedCacheKeys.kt index af5f1eb0..863bbba9 100644 --- a/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedCacheKeys.kt +++ b/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedCacheKeys.kt @@ -9,7 +9,7 @@ object SubscriptionFeedCacheKeys { fun invalidation(userId: String): String = "feed:invalidation:${hash(userId)}" - fun selection(userId: String, token: String): String = "feed:selection:${hash(userId)}:$token" + fun selection(userId: String, slot: Int): String = "feed:selection:${hash(userId)}:$slot" fun shorts(userId: String): String = "feed:shorts:${hash(userId)}" diff --git a/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedSelectionStore.kt b/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedSelectionStore.kt index 27d42252..b8bc1737 100644 --- a/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedSelectionStore.kt +++ b/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedSelectionStore.kt @@ -3,7 +3,7 @@ package dev.typetype.server.services import dev.typetype.server.cache.CacheJson import dev.typetype.server.cache.CacheService import kotlinx.serialization.Serializable -import java.util.UUID +import java.security.MessageDigest internal class SubscriptionFeedSelectionStore( private val cache: CacheService, @@ -17,29 +17,83 @@ internal class SubscriptionFeedSelectionStore( if (selection == SubscriptionSelection.All) return SubscriptionFeedSelectionSnapshot(null, null) if (token == null) { val channelUrls = subscriptions.getChannelUrls(userId, selection) - val nextToken = UUID.randomUUID().toString() - cache.set( - SubscriptionFeedCacheKeys.selection(userId, nextToken), - CacheJson.encodeToString( - StoredSubscriptionFeedSelection.serializer(), - StoredSubscriptionFeedSelection(selection.cursorKey, channelUrls.toList()), - ), - SubscriptionFeedSnapshotStore.RETENTION_SECONDS, + return SubscriptionFeedSelectionSnapshot( + token = tokenFor(selection.cursorKey, channelUrls), + channelUrls = channelUrls, ) - return SubscriptionFeedSelectionSnapshot(nextToken, channelUrls) } - val raw = runCatching { cache.get(SubscriptionFeedCacheKeys.selection(userId, token)) }.getOrNull() + for (slot in 0 until MAX_SNAPSHOTS_PER_USER) { + val stored = read(userId, slot) ?: continue + if (stored.token != token) continue + if (stored.filterKey != selection.cursorKey) return null + return SubscriptionFeedSelectionSnapshot(token, stored.channelUrls.toSet()) + } + return null + } + + suspend fun persist( + userId: String, + selection: SubscriptionSelection, + snapshot: SubscriptionFeedSelectionSnapshot, + ): Boolean { + val token = snapshot.token ?: return true + val channelUrls = snapshot.channelUrls ?: return true + val stored = StoredSubscriptionFeedSelection( + token = token, + filterKey = selection.cursorKey, + channelUrls = channelUrls.sorted(), + ) + val encoded = CacheJson.encodeToString(StoredSubscriptionFeedSelection.serializer(), stored) + for (slot in 0 until MAX_SNAPSHOTS_PER_USER) { + val current = read(userId, slot) + if (current != null) { + if (current == stored) { + val key = SubscriptionFeedCacheKeys.selection(userId, slot) + if (cache.refreshIfValueMatches(key, encoded, SubscriptionFeedSnapshotStore.RETENTION_SECONDS)) { + return true + } + } + continue + } + val key = SubscriptionFeedCacheKeys.selection(userId, slot) + if (cache.setIfAbsent(key, encoded, SubscriptionFeedSnapshotStore.RETENTION_SECONDS)) return true + if (read(userId, slot) == stored) return true + } + return false + } + + private suspend fun read(userId: String, slot: Int): StoredSubscriptionFeedSelection? { + val raw = runCatching { cache.get(SubscriptionFeedCacheKeys.selection(userId, slot)) }.getOrNull() ?: return null - val stored = runCatching { + return runCatching { CacheJson.decodeFromString(StoredSubscriptionFeedSelection.serializer(), raw) - }.getOrNull() ?: return null - if (stored.filterKey != selection.cursorKey) return null - return SubscriptionFeedSelectionSnapshot(token, stored.channelUrls.toSet()) + }.getOrNull() + } + + private fun tokenFor(filterKey: String, channelUrls: Set): String { + val identity = CacheJson.encodeToString( + SubscriptionFeedSelectionIdentity.serializer(), + SubscriptionFeedSelectionIdentity(filterKey, channelUrls.sorted()), + ) + return MessageDigest.getInstance("SHA-256") + .digest(identity.toByteArray()) + .joinToString("") { "%02x".format(it) } + } + + private companion object { + const val MAX_SNAPSHOTS_PER_USER = 8 } } @Serializable private data class StoredSubscriptionFeedSelection( + val token: String, + val filterKey: String, + val channelUrls: List, +) + +@Serializable +private data class SubscriptionFeedSelectionIdentity( val filterKey: String, val channelUrls: List, ) diff --git a/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedService.kt b/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedService.kt index 38e75d8d..2bef39f8 100644 --- a/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedService.kt +++ b/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedService.kt @@ -71,18 +71,20 @@ class SubscriptionFeedService( val offset = cursorState?.offset ?: page * limit val selected = selections.resolve(userId, selection, cursorState?.selectionToken) ?: return SubscriptionFeedPageResult.StaleGeneration - return SubscriptionFeedPageResult.Ready( - snapshot.page( - offset, - limit, - isRefreshing(userId), - hideLiveStreams, - hideMembersOnlyContent, - selection, - selected.channelUrls, - selected.token, - ), + val response = snapshot.page( + offset, + limit, + isRefreshing(userId), + hideLiveStreams, + hideMembersOnlyContent, + selection, + selected.channelUrls, + selected.token, ) + if (cursorState == null && response.nextpage != null && !selections.persist(userId, selection, selected)) { + return SubscriptionFeedPageResult.CursorCapacityReached + } + return SubscriptionFeedPageResult.Ready(response) } suspend fun getFeed(userId: String, page: Int, limit: Int): SubscriptionFeedResponse = diff --git a/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedSnapshot.kt b/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedSnapshot.kt index 5d8232a1..e724cf0d 100644 --- a/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedSnapshot.kt +++ b/src/main/kotlin/dev/typetype/server/services/SubscriptionFeedSnapshot.kt @@ -51,7 +51,8 @@ internal object SubscriptionFeedCursorCodec { val cursor = CacheJson.decodeFromString(SubscriptionFeedCursor.serializer(), payload) cursor.takeIf { it.generation > 0L && it.offset >= 0 && it.limit in 1..100 && - ((it.filterKey == SubscriptionSelection.All.cursorKey) == (it.selectionToken == null)) + ((it.filterKey == SubscriptionSelection.All.cursorKey) == (it.selectionToken == null)) && + (it.selectionToken == null || SELECTION_TOKEN.matches(it.selectionToken)) } ?.let { SubscriptionFeedCursorState( @@ -65,6 +66,8 @@ internal object SubscriptionFeedCursorCodec { ) } }.getOrNull() + + private val SELECTION_TOKEN = Regex("[0-9a-f]{64}") } internal data class SubscriptionFeedCursorState( @@ -137,4 +140,5 @@ internal sealed interface SubscriptionFeedPageResult { data class Preparing(val retryAfterMs: Long) : SubscriptionFeedPageResult data object InvalidCursor : SubscriptionFeedPageResult data object StaleGeneration : SubscriptionFeedPageResult + data object CursorCapacityReached : SubscriptionFeedPageResult } diff --git a/src/test/kotlin/dev/typetype/server/FakeCacheService.kt b/src/test/kotlin/dev/typetype/server/FakeCacheService.kt index ac2319ec..d046ba93 100644 --- a/src/test/kotlin/dev/typetype/server/FakeCacheService.kt +++ b/src/test/kotlin/dev/typetype/server/FakeCacheService.kt @@ -12,6 +12,18 @@ class FakeCacheService : CacheService { values[key] = value } + override suspend fun setIfAbsent(key: String, value: String, ttlSeconds: Long): Boolean = + values.putIfAbsent(key, value) == null + + override suspend fun refreshIfValueMatches(key: String, value: String, ttlSeconds: Long): Boolean { + var matched = false + values.computeIfPresent(key) { _, current -> + matched = current == value + current + } + return matched + } + override suspend fun delete(key: String) { values.remove(key) } @@ -19,4 +31,6 @@ class FakeCacheService : CacheService { fun clear() { values.clear() } + + fun keys(): Set = values.keys.toSet() } diff --git a/src/test/kotlin/dev/typetype/server/SubscriptionFeedSelectionStoreTest.kt b/src/test/kotlin/dev/typetype/server/SubscriptionFeedSelectionStoreTest.kt new file mode 100644 index 00000000..d528297f --- /dev/null +++ b/src/test/kotlin/dev/typetype/server/SubscriptionFeedSelectionStoreTest.kt @@ -0,0 +1,73 @@ +package dev.typetype.server + +import dev.typetype.server.services.SubscriptionFeedSelectionSnapshot +import dev.typetype.server.services.SubscriptionFeedSelectionStore +import dev.typetype.server.services.SubscriptionSelection +import dev.typetype.server.services.SubscriptionsService +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.test.runTest +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertNotNull +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test + +class SubscriptionFeedSelectionStoreTest { + @Test + fun `a ninth distinct session cannot evict the first issued cursor`() = runTest { + val cache = FakeCacheService() + val store = SubscriptionFeedSelectionStore(cache, SubscriptionsService()) + val selections = (1..9).map { index -> + val selection = SubscriptionSelection.Group("group-$index") + val snapshot = SubscriptionFeedSelectionSnapshot( + token = index.toString().padStart(64, '0'), + channelUrls = setOf("https://example.com/channel/$index"), + ) + selection to snapshot + } + + selections.take(8).forEach { (selection, snapshot) -> + assertTrue(store.persist(TEST_USER_ID, selection, snapshot)) + } + val (ninthSelection, ninthSnapshot) = selections.last() + assertFalse(store.persist(TEST_USER_ID, ninthSelection, ninthSnapshot)) + + val (firstSelection, firstSnapshot) = selections.first() + val restored = store.resolve(TEST_USER_ID, firstSelection, firstSnapshot.token) + assertNotNull(restored) + assertEquals(firstSnapshot.channelUrls, restored?.channelUrls) + assertEquals(8, cache.keys().count { it.startsWith("feed:selection") }) + } + + @Test + fun `independent store instances cannot overwrite concurrently issued cursors`() = runTest { + val cache = FakeCacheService() + val stores = List(2) { SubscriptionFeedSelectionStore(cache, SubscriptionsService()) } + val selections = List(2) { index -> + val number = index + 1 + val selection = SubscriptionSelection.Group("group-$number") + val snapshot = SubscriptionFeedSelectionSnapshot( + token = number.toString().padStart(64, '0'), + channelUrls = setOf("https://example.com/channel/$number"), + ) + selection to snapshot + } + val start = CompletableDeferred() + val writes = stores.zip(selections).map { (store, pair) -> + async(Dispatchers.Default) { + start.await() + store.persist(TEST_USER_ID, pair.first, pair.second) + } + } + + start.complete(Unit) + + assertTrue(writes.awaitAll().all { it }) + selections.forEachIndexed { index, (selection, snapshot) -> + assertNotNull(stores[index].resolve(TEST_USER_ID, selection, snapshot.token)) + } + } +} diff --git a/src/test/kotlin/dev/typetype/server/SubscriptionGroupFeedRoutesTest.kt b/src/test/kotlin/dev/typetype/server/SubscriptionGroupFeedRoutesTest.kt index c2becbaa..e4c88e8c 100644 --- a/src/test/kotlin/dev/typetype/server/SubscriptionGroupFeedRoutesTest.kt +++ b/src/test/kotlin/dev/typetype/server/SubscriptionGroupFeedRoutesTest.kt @@ -26,6 +26,7 @@ import io.mockk.coEvery import io.mockk.mockk import kotlinx.serialization.json.Json import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNotEquals import org.junit.jupiter.api.Assertions.assertTrue import org.junit.jupiter.api.BeforeAll import org.junit.jupiter.api.BeforeEach @@ -35,6 +36,7 @@ class SubscriptionGroupFeedRoutesTest { private val subscriptions = SubscriptionsService() private val groups = SubscriptionGroupsService() private lateinit var feed: SubscriptionFeedService + private lateinit var cache: FakeCacheService private val auth = AuthService.fixed(TEST_USER_ID) companion object { @@ -46,7 +48,8 @@ class SubscriptionGroupFeedRoutesTest { @BeforeEach fun clean() { TestDatabase.truncateAll() - feed = SubscriptionFeedService(subscriptions, FakeChannelService(), FakeCacheService()) + cache = FakeCacheService() + feed = SubscriptionFeedService(subscriptions, FakeChannelService(), cache) } private fun withApp(block: suspend ApplicationTestBuilder.() -> Unit) = testApplication { @@ -87,7 +90,7 @@ class SubscriptionGroupFeedRoutesTest { coEvery { channelService.getChannel(channel("three"), null) } returns SubscriptionFeedTestFixtures.channel( SubscriptionFeedTestFixtures.video(1_000L, channel = "three", url = "video-three"), ) - feed = SubscriptionFeedService(subscriptions, channelService, FakeCacheService()) + feed = SubscriptionFeedService(subscriptions, channelService, cache) listOf("one", "two", "three").forEach { subscriptions.add(TEST_USER_ID, subscription(it)) } val group = (groups.create(TEST_USER_ID, "Work") as SubscriptionGroupWriteResult.Success).group groups.addSubscription(TEST_USER_ID, group.id, channel("one")) @@ -96,15 +99,34 @@ class SubscriptionGroupFeedRoutesTest { feed.awaitRefresh(TEST_USER_ID) val firstPage = requestReadyFeed(limit = 1, groupId = group.id) assertEquals(listOf("video-one"), firstPage.videos.map { it.url }) + val repeatedFirstPage = requestReadyFeed(limit = 1, groupId = group.id) + assertEquals(firstPage.nextpage, repeatedFirstPage.nextpage) + assertEquals(1, cache.keys().count { it.startsWith("feed:selection") }) groups.removeSubscription(TEST_USER_ID, group.id, channel("two")) groups.addSubscription(TEST_USER_ID, group.id, channel("three")) + val changedFirstPage = requestReadyFeed(limit = 1, groupId = group.id) + assertNotEquals(firstPage.nextpage, changedFirstPage.nextpage) + assertEquals(2, cache.keys().count { it.startsWith("feed:selection") }) val secondPage = requestFeed(limit = 1, cursor = requireNotNull(firstPage.nextpage), groupId = group.id) assertEquals(HttpStatusCode.OK, secondPage.status) assertEquals(listOf("video-two"), Json.decodeFromString(secondPage.bodyAsText()).videos.map { it.url }) } + @Test + fun `terminal filtered page does not retain a membership snapshot`() = withApp { + subscriptions.add(TEST_USER_ID, subscription("one")) + val group = (groups.create(TEST_USER_ID, "Work") as SubscriptionGroupWriteResult.Success).group + groups.addSubscription(TEST_USER_ID, group.id, channel("one")) + assertEquals(HttpStatusCode.Accepted, requestFeed(groupId = group.id).status) + feed.awaitRefresh(TEST_USER_ID) + + assertEquals(1, requestReadyFeed(groupId = group.id).videos.size) + + assertTrue(cache.keys().none { it.startsWith("feed:selection") }) + } + @Test fun `cursor cannot be reused with another subscription filter`() = withApp { subscriptions.add(TEST_USER_ID, subscription("one")) diff --git a/src/test/kotlin/dev/typetype/server/SubscriptionsRoutesTest.kt b/src/test/kotlin/dev/typetype/server/SubscriptionsRoutesTest.kt index df2da9ac..96097f40 100644 --- a/src/test/kotlin/dev/typetype/server/SubscriptionsRoutesTest.kt +++ b/src/test/kotlin/dev/typetype/server/SubscriptionsRoutesTest.kt @@ -19,14 +19,12 @@ import io.ktor.server.plugins.contentnegotiation.ContentNegotiation import io.ktor.server.routing.routing import io.ktor.server.testing.ApplicationTestBuilder import io.ktor.server.testing.testApplication +import kotlinx.serialization.json.Json import org.junit.jupiter.api.Assertions.assertEquals -import org.junit.jupiter.api.Assertions.assertFalse import org.junit.jupiter.api.Assertions.assertTrue import org.junit.jupiter.api.BeforeAll import org.junit.jupiter.api.BeforeEach import org.junit.jupiter.api.Test -import java.nio.file.Files -import java.nio.file.Path class SubscriptionsRoutesTest { @@ -44,7 +42,7 @@ class SubscriptionsRoutesTest { private fun withApp(block: suspend ApplicationTestBuilder.() -> Unit) = testApplication { application { - install(ContentNegotiation) { json() } + install(ContentNegotiation) { json(Json { ignoreUnknownKeys = true; encodeDefaults = true }) } routing { subscriptionsRoutes(service, auth) } } block() @@ -52,19 +50,6 @@ class SubscriptionsRoutesTest { private val itemBody = """{"channelUrl":"https://yt.com/channel/1","name":"Test","avatarUrl":""}""" - @Test - fun `subscription creation contract omits the server timestamp`() { - val components = Files.readString(Path.of("openapi/components/subscriptions.yaml")) - val requestSchema = components - .substringAfter("SubscriptionCreateRequest:") - .substringBefore("SubscriptionGroupItem:") - val paths = Files.readString(Path.of("openapi/paths/subscriptions.yaml")) - - assertTrue("required: [channelUrl, name, avatarUrl]" in requestSchema) - assertFalse("subscribedAt" in requestSchema) - assertTrue("#/SubscriptionCreateRequest" in paths) - } - @Test fun `GET subscriptions without token returns 401`() = withApp { assertEquals(HttpStatusCode.Unauthorized, client.get("/subscriptions").status) @@ -78,14 +63,31 @@ class SubscriptionsRoutesTest { } @Test - fun `POST subscriptions returns 201 and persists item`() = withApp { + fun `POST subscriptions generates and persists the server timestamp`() = withApp { val response = client.post("/subscriptions") { headers.append(HttpHeaders.Authorization, "Bearer test-jwt") headers.append(HttpHeaders.ContentType, ContentType.Application.Json.toString()) setBody(itemBody) } assertEquals(HttpStatusCode.Created, response.status) - assertTrue(response.bodyAsText().contains("\"channelUrl\":\"https://yt.com/channel/1\"")) + val created = Json.decodeFromString(response.bodyAsText()) + assertEquals("https://yt.com/channel/1", created.channelUrl) + assertTrue(created.subscribedAt > 1) + assertEquals(created.subscribedAt, service.getAll(TEST_USER_ID).single().subscribedAt) + } + + @Test + fun `POST subscriptions ignores the obsolete client timestamp`() = withApp { + val response = client.post("/subscriptions") { + headers.append(HttpHeaders.Authorization, "Bearer test-jwt") + headers.append(HttpHeaders.ContentType, ContentType.Application.Json.toString()) + setBody(itemBody.dropLast(1) + ",\"subscribedAt\":1}") + } + + assertEquals(HttpStatusCode.Created, response.status) + val created = Json.decodeFromString(response.bodyAsText()) + assertTrue(created.subscribedAt > 1) + assertEquals(created.subscribedAt, service.getAll(TEST_USER_ID).single().subscribedAt) } @Test diff --git a/src/test/kotlin/dev/typetype/server/TypeTypeBackupServiceTest.kt b/src/test/kotlin/dev/typetype/server/TypeTypeBackupServiceTest.kt index 7d9a2219..1d9a7d8f 100644 --- a/src/test/kotlin/dev/typetype/server/TypeTypeBackupServiceTest.kt +++ b/src/test/kotlin/dev/typetype/server/TypeTypeBackupServiceTest.kt @@ -152,6 +152,31 @@ class TypeTypeBackupServiceTest { assertTrue(backup.history == null) } + @Test + fun `legacy subscription backup without groups preserves compatible memberships`() = runTest { + subscriptions.add(TARGET, SubscriptionItem("https://youtube.com/channel/keep", "Keep", "")) + subscriptions.add(TARGET, SubscriptionItem("https://youtube.com/channel/drop", "Drop", "")) + val group = (subscriptionGroups.create(TARGET, "Existing") as SubscriptionGroupWriteResult.Success).group + subscriptionGroups.addSubscription(TARGET, group.id, "https://youtube.com/channel/keep") + subscriptionGroups.addSubscription(TARGET, group.id, "https://youtube.com/channel/drop") + val legacyBackup = TypeTypeBackupItem( + exportedAt = 1, + categories = listOf(TypeTypeBackupCategory.SUBSCRIPTIONS.wireName), + subscriptions = listOf( + SubscriptionItem("https://youtube.com/channel/keep", "Keep", "", subscribedAt = 1), + ), + ) + + service.restore(TARGET, legacyBackup) + + val preservedGroup = subscriptionGroups.getAll(TARGET).single() + assertEquals(group.id, preservedGroup.id) + assertEquals( + listOf("https://youtube.com/channel/keep"), + subscriptionGroups.getChannelUrls(TARGET, preservedGroup.id), + ) + } + @Test fun `restore rejects empty normalized blocked keywords`() = runTest { val backup = TypeTypeBackupItem( From 93a0830aae597477c51c85f9248fdde93c350aab Mon Sep 17 00:00:00 2001 From: User Date: Mon, 17 Aug 2026 19:46:54 -0700 Subject: [PATCH 12/14] fix: prevent orphaned subscription memberships Subscription group assignment checked subscription ownership without coordinating with unsubscribe or replacement restores. Take an account-keyed transaction advisory lock across membership assignment and every subscription removal or replacement path so the check and insert cannot straddle a committed deletion. Constraint: Replacement imports must retain memberships whose subscriptions survive the import Rejected: Composite foreign key with cascading deletes | cascade semantics would discard memberships before retained subscriptions are reinserted Confidence: high Scope-risk: narrow Directive: Any new path that removes or replaces an account's subscriptions must acquire SubscriptionMutationLock in the same transaction Tested: Focused PostgreSQL concurrency regression on JDK 25 Not-tested: Full suite and live HTTP concurrency gate run after both fix commits --- .../PipePipeBackupPersisterService.kt | 1 + .../services/SubscriptionGroupsService.kt | 1 + .../services/SubscriptionMutationLock.kt | 14 +++ .../server/services/SubscriptionsService.kt | 1 + .../services/TypeTypeBackupRestoreWriter.kt | 1 + .../server/SubscriptionGroupsServiceTest.kt | 89 +++++++++++++++++++ 6 files changed, 107 insertions(+) create mode 100644 src/main/kotlin/dev/typetype/server/services/SubscriptionMutationLock.kt diff --git a/src/main/kotlin/dev/typetype/server/services/PipePipeBackupPersisterService.kt b/src/main/kotlin/dev/typetype/server/services/PipePipeBackupPersisterService.kt index 4e965014..4ee60d97 100644 --- a/src/main/kotlin/dev/typetype/server/services/PipePipeBackupPersisterService.kt +++ b/src/main/kotlin/dev/typetype/server/services/PipePipeBackupPersisterService.kt @@ -16,6 +16,7 @@ import java.util.UUID class PipePipeBackupPersisterService { suspend fun persist(userId: String, snapshot: PipePipeBackupSnapshotItem): PipePipeBackupRestoreResult = DatabaseFactory.query { + SubscriptionMutationLock.acquire(userId) clearUserData(userId) val avatarsByChannel = snapshot.subscriptions .mapNotNull { item -> item.url.takeIf { it.isNotBlank() }?.let { url -> url to item.avatarUrl } } diff --git a/src/main/kotlin/dev/typetype/server/services/SubscriptionGroupsService.kt b/src/main/kotlin/dev/typetype/server/services/SubscriptionGroupsService.kt index 23887b2c..2566c5ed 100644 --- a/src/main/kotlin/dev/typetype/server/services/SubscriptionGroupsService.kt +++ b/src/main/kotlin/dev/typetype/server/services/SubscriptionGroupsService.kt @@ -101,6 +101,7 @@ class SubscriptionGroupsService { groupId: String, rawChannelUrl: String, ): SubscriptionGroupMembershipResult = DatabaseFactory.query { + SubscriptionMutationLock.acquire(userId) if (!groupExists(userId, groupId)) return@query SubscriptionGroupMembershipResult.GroupNotFound val channelUrl = ChannelUrlCanonicalizer.canonicalize(rawChannelUrl) val subscriptionExists = SubscriptionsTable.selectAll().where { diff --git a/src/main/kotlin/dev/typetype/server/services/SubscriptionMutationLock.kt b/src/main/kotlin/dev/typetype/server/services/SubscriptionMutationLock.kt new file mode 100644 index 00000000..65caa657 --- /dev/null +++ b/src/main/kotlin/dev/typetype/server/services/SubscriptionMutationLock.kt @@ -0,0 +1,14 @@ +package dev.typetype.server.services + +import org.jetbrains.exposed.v1.jdbc.transactions.TransactionManager + +internal object SubscriptionMutationLock { + fun acquire(userId: String) { + val userKey = userId.hashCode() and Int.MAX_VALUE + TransactionManager.current().exec( + "SELECT pg_advisory_xact_lock($LOCK_NAMESPACE, $userKey)", + ) + } + + private const val LOCK_NAMESPACE = 1_414_814_032 +} diff --git a/src/main/kotlin/dev/typetype/server/services/SubscriptionsService.kt b/src/main/kotlin/dev/typetype/server/services/SubscriptionsService.kt index 21d022de..6e90ab8f 100644 --- a/src/main/kotlin/dev/typetype/server/services/SubscriptionsService.kt +++ b/src/main/kotlin/dev/typetype/server/services/SubscriptionsService.kt @@ -46,6 +46,7 @@ class SubscriptionsService { } suspend fun delete(userId: String, channelUrl: String): Boolean = DatabaseFactory.query { + SubscriptionMutationLock.acquire(userId) val canonicalUrl = ChannelUrlCanonicalizer.canonicalize(channelUrl) SubscriptionGroupMembershipsTable.deleteWhere { (SubscriptionGroupMembershipsTable.userId eq userId) and diff --git a/src/main/kotlin/dev/typetype/server/services/TypeTypeBackupRestoreWriter.kt b/src/main/kotlin/dev/typetype/server/services/TypeTypeBackupRestoreWriter.kt index d951dafd..b8209cb9 100644 --- a/src/main/kotlin/dev/typetype/server/services/TypeTypeBackupRestoreWriter.kt +++ b/src/main/kotlin/dev/typetype/server/services/TypeTypeBackupRestoreWriter.kt @@ -10,6 +10,7 @@ internal object TypeTypeBackupRestoreWriter { backup: TypeTypeBackupItem, categories: Set, ): TypeTypeRestoreSummary = DatabaseFactory.query { + SubscriptionMutationLock.acquire(userId) val restored = linkedMapOf() if (TypeTypeBackupCategory.SUBSCRIPTIONS in categories) { restored["subscriptions"] = TypeTypeBackupCoreRestore.subscriptions( diff --git a/src/test/kotlin/dev/typetype/server/SubscriptionGroupsServiceTest.kt b/src/test/kotlin/dev/typetype/server/SubscriptionGroupsServiceTest.kt index d03bce44..74ba5762 100644 --- a/src/test/kotlin/dev/typetype/server/SubscriptionGroupsServiceTest.kt +++ b/src/test/kotlin/dev/typetype/server/SubscriptionGroupsServiceTest.kt @@ -1,6 +1,7 @@ package dev.typetype.server import dev.typetype.server.models.SubscriptionItem +import dev.typetype.server.models.TypeTypeBackupItem import dev.typetype.server.db.DatabaseFactory import dev.typetype.server.services.SubscriptionGroupMembershipCleaner import dev.typetype.server.services.SubscriptionGroupMembershipResult @@ -8,13 +9,23 @@ import dev.typetype.server.services.SubscriptionGroupWriteResult import dev.typetype.server.services.SubscriptionGroupsService import dev.typetype.server.services.SubscriptionSelection import dev.typetype.server.services.SubscriptionsService +import dev.typetype.server.services.TypeTypeBackupCategory +import dev.typetype.server.services.TypeTypeBackupRestoreWriter +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.async import kotlinx.coroutines.test.runTest +import kotlinx.coroutines.withContext +import kotlinx.coroutines.withTimeoutOrNull +import kotlinx.coroutines.yield +import org.jetbrains.exposed.v1.jdbc.transactions.TransactionManager import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Assertions.assertFalse import org.junit.jupiter.api.Assertions.assertTrue import org.junit.jupiter.api.BeforeAll import org.junit.jupiter.api.BeforeEach import org.junit.jupiter.api.Test +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit class SubscriptionGroupsServiceTest { private val groups = SubscriptionGroupsService() @@ -98,6 +109,61 @@ class SubscriptionGroupsServiceTest { assertEquals(emptyList(), groups.getChannelUrls("user", group.id)) } + @Test + fun `membership assignment deletion and replacement share a user lock`() = runTest { + val userId = "concurrent-user" + val group = groups.create(userId, "Group").createdGroup() + subscriptions.add(userId, subscription("one")) + val lockHeld = CountDownLatch(1) + val releaseLock = CountDownLatch(1) + val holder = async(Dispatchers.IO) { + DatabaseFactory.query { + TransactionManager.current().exec(subscriptionLockSql(userId)) + lockHeld.countDown() + check(releaseLock.await(5, TimeUnit.SECONDS)) + } + } + assertTrue(lockHeld.await(5, TimeUnit.SECONDS)) + + val assignment = async(Dispatchers.IO) { + groups.addSubscription(userId, group.id, channel("one")) + } + val deletion = async(Dispatchers.IO) { subscriptions.delete(userId, channel("one")) } + val replacement = async(Dispatchers.IO) { + TypeTypeBackupRestoreWriter.restore( + userId = userId, + backup = TypeTypeBackupItem( + exportedAt = 1, + categories = listOf(TypeTypeBackupCategory.SUBSCRIPTIONS.wireName), + subscriptions = listOf(subscription("one").copy(subscribedAt = 1)), + ), + categories = setOf(TypeTypeBackupCategory.SUBSCRIPTIONS), + ) + } + val allWaited = try { + withContext(Dispatchers.IO) { + withTimeoutOrNull(2_000L) { + var waiting = false + while (!waiting && !(assignment.isCompleted && deletion.isCompleted && replacement.isCompleted)) { + waiting = waitingSubscriptionLocks(userId) >= 3 + if (!waiting) yield() + } + waiting + } ?: false + } + } finally { + releaseLock.countDown() + } + + holder.await() + assignment.await() + assertTrue(deletion.await()) + replacement.await() + assertTrue(allWaited, "all mutations must wait for the same account-scoped lock") + val subscriptionUrls = subscriptions.getAll(userId).mapTo(hashSetOf(), SubscriptionItem::channelUrl) + assertTrue(groups.getChannelUrls(userId, group.id).all { it in subscriptionUrls }) + } + @Test fun `replacement imports retain only memberships for subscriptions still present`() = runTest { val group = groups.create("user", "Group").createdGroup() @@ -117,4 +183,27 @@ class SubscriptionGroupsServiceTest { private fun subscription(id: String) = SubscriptionItem(channel(id), id, "") private fun channel(id: String) = "https://yt.com/channel/$id" + + private fun subscriptionLockSql(userId: String): String = + "SELECT pg_advisory_xact_lock($SUBSCRIPTION_LOCK_NAMESPACE, ${subscriptionLockKey(userId)})" + + private suspend fun waitingSubscriptionLocks(userId: String): Int = DatabaseFactory.query { + TransactionManager.current().exec( + """ + SELECT count(*) + FROM pg_locks + WHERE locktype = 'advisory' + AND classid = $SUBSCRIPTION_LOCK_NAMESPACE + AND objid = ${subscriptionLockKey(userId)} + AND NOT granted + """.trimIndent(), + ) { result -> + result.next() + result.getInt(1) + } ?: 0 + } + + private fun subscriptionLockKey(userId: String): Int = userId.hashCode() and Int.MAX_VALUE } + +private const val SUBSCRIPTION_LOCK_NAMESPACE = 1_414_814_032 From 267fda9a7cf7854655b4289f9cf7665c33f26c2f Mon Sep 17 00:00:00 2001 From: User Date: Mon, 17 Aug 2026 19:47:08 -0700 Subject: [PATCH 13/14] fix: keep exported subscription backups restorable Subscription and group sections were read in separate transactions, so a membership committed between them could reference a subscription absent from the exported list. Capture subscriptions once and export only group memberships belonging to that captured set, preserving the restore validator's referential invariant. Constraint: Subscription groups remain coupled to the subscriptions backup category Rejected: Add a cross-service export transaction | existing service reads open their own transactions and captured-set filtering is the smaller accepted repair Confidence: high Scope-risk: narrow Directive: Exported group memberships must remain a subset of the subscriptions captured for the same backup Tested: Focused mixed-read export and restore regression on JDK 25 Not-tested: Full suite and live HTTP concurrency gate run after this commit --- .../SubscriptionGroupBackupRepository.kt | 8 ++- .../server/services/TypeTypeBackupService.kt | 16 +++-- .../SubscriptionBackupConsistencyTest.kt | 72 +++++++++++++++++++ 3 files changed, 90 insertions(+), 6 deletions(-) create mode 100644 src/test/kotlin/dev/typetype/server/SubscriptionBackupConsistencyTest.kt diff --git a/src/main/kotlin/dev/typetype/server/services/SubscriptionGroupBackupRepository.kt b/src/main/kotlin/dev/typetype/server/services/SubscriptionGroupBackupRepository.kt index 3cc685aa..3df60b25 100644 --- a/src/main/kotlin/dev/typetype/server/services/SubscriptionGroupBackupRepository.kt +++ b/src/main/kotlin/dev/typetype/server/services/SubscriptionGroupBackupRepository.kt @@ -13,10 +13,16 @@ import java.util.Locale import java.util.UUID internal object SubscriptionGroupBackupRepository { - suspend fun export(userId: String): List = DatabaseFactory.query { + suspend fun export( + userId: String, + subscriptionUrls: Set, + ): List = DatabaseFactory.query { val channelsByGroup = SubscriptionGroupMembershipsTable.selectAll() .where { SubscriptionGroupMembershipsTable.userId eq userId } .orderBy(SubscriptionGroupMembershipsTable.addedAt to SortOrder.ASC) + .filter { + ChannelUrlCanonicalizer.canonicalize(it[SubscriptionGroupMembershipsTable.channelUrl]) in subscriptionUrls + } .groupBy( keySelector = { it[SubscriptionGroupMembershipsTable.groupId] }, valueTransform = { it[SubscriptionGroupMembershipsTable.channelUrl] }, diff --git a/src/main/kotlin/dev/typetype/server/services/TypeTypeBackupService.kt b/src/main/kotlin/dev/typetype/server/services/TypeTypeBackupService.kt index 97466980..fb341ae2 100644 --- a/src/main/kotlin/dev/typetype/server/services/TypeTypeBackupService.kt +++ b/src/main/kotlin/dev/typetype/server/services/TypeTypeBackupService.kt @@ -31,14 +31,20 @@ class TypeTypeBackupService( } else { null } + val subscriptionItems = if (includes(TypeTypeBackupCategory.SUBSCRIPTIONS)) { + subscriptions.getAll(userId) + } else { + null + } return TypeTypeBackupItem( exportedAt = System.currentTimeMillis(), categories = categories.map(TypeTypeBackupCategory::wireName).sorted(), - subscriptions = if (includes(TypeTypeBackupCategory.SUBSCRIPTIONS)) subscriptions.getAll(userId) else null, - subscriptionGroups = if (includes(TypeTypeBackupCategory.SUBSCRIPTIONS)) { - SubscriptionGroupBackupRepository.export(userId) - } else { - null + subscriptions = subscriptionItems, + subscriptionGroups = subscriptionItems?.let { items -> + val channelUrls = items.mapTo(hashSetOf()) { + ChannelUrlCanonicalizer.canonicalize(it.channelUrl) + } + SubscriptionGroupBackupRepository.export(userId, channelUrls) }, history = if (includes(TypeTypeBackupCategory.HISTORY)) history.getAll(userId) else null, playlists = fullPlaylists, diff --git a/src/test/kotlin/dev/typetype/server/SubscriptionBackupConsistencyTest.kt b/src/test/kotlin/dev/typetype/server/SubscriptionBackupConsistencyTest.kt new file mode 100644 index 00000000..02afdec5 --- /dev/null +++ b/src/test/kotlin/dev/typetype/server/SubscriptionBackupConsistencyTest.kt @@ -0,0 +1,72 @@ +package dev.typetype.server + +import dev.typetype.server.models.SubscriptionItem +import dev.typetype.server.services.SubscriptionGroupMembershipResult +import dev.typetype.server.services.SubscriptionGroupsService +import dev.typetype.server.services.SubscriptionGroupWriteResult +import dev.typetype.server.services.SubscriptionsService +import dev.typetype.server.services.TypeTypeBackupCategory +import dev.typetype.server.services.TypeTypeBackupService +import io.mockk.coEvery +import io.mockk.mockk +import kotlinx.coroutines.test.runTest +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.BeforeAll +import org.junit.jupiter.api.BeforeEach +import org.junit.jupiter.api.Test + +class SubscriptionBackupConsistencyTest { + private val subscriptions = SubscriptionsService() + private val groups = SubscriptionGroupsService() + + companion object { + @BeforeAll + @JvmStatic + fun initDb() = TestDatabase.setup() + } + + @BeforeEach + fun clean() = TestDatabase.truncateAll() + + @Test + fun `backup stays restorable when a subscription is added between section reads`() = runTest { + val group = (groups.create(SOURCE, "New") as SubscriptionGroupWriteResult.Success).group + val capturedSubscriptions = mockk() + coEvery { capturedSubscriptions.getAll(SOURCE, any()) } coAnswers { + subscriptions.add(SOURCE, SubscriptionItem(CHANNEL_URL, "Channel", "")) + assertEquals( + SubscriptionGroupMembershipResult.Success, + groups.addSubscription(SOURCE, group.id, CHANNEL_URL), + ) + emptyList() + } + val service = backupService(capturedSubscriptions) + + val backup = service.export(SOURCE, setOf(TypeTypeBackupCategory.SUBSCRIPTIONS)) + val restored = service.restore(TARGET, backup) + + assertEquals(emptyList(), backup.subscriptions) + assertEquals(emptyList(), backup.subscriptionGroups?.single()?.channelUrls) + assertEquals(1, restored.restored["subscriptionGroups"]) + assertEquals(0, restored.restored["subscriptionGroupMemberships"]) + } + + private fun backupService(subscriptions: SubscriptionsService) = TypeTypeBackupService( + subscriptions = subscriptions, + history = mockk(), + playlists = mockk(), + watchLater = mockk(), + favorites = mockk(), + progress = mockk(), + searchHistory = mockk(), + savedPlaylists = mockk(), + settings = mockk(), + blocked = mockk(), + allowedChannels = mockk(), + allowedPlaylists = mockk(), + ) +} + +private const val SOURCE = "concurrent-backup-source" +private const val TARGET = "concurrent-backup-target" +private const val CHANNEL_URL = "https://youtube.com/channel/concurrent" From af9fb57b8ee108071f45ae775562bef2d2f7839c Mon Sep 17 00:00:00 2001 From: User Date: Mon, 17 Aug 2026 22:01:16 -0700 Subject: [PATCH 14/14] refactor: clarify subscription lock namespace The previous decimal constant encoded an undocumented product tag. Use the precomputed PostgreSQL hashtext value for the literal subscriptions namespace and document its origin while retaining numeric lock lookup at runtime. Constraint: Do not evaluate hashtext on every lock acquisition Rejected: Runtime hashtext('subscriptions') | repeats derivation on every lock lookup Confidence: high Scope-risk: narrow Directive: Keep the production and concurrency-test namespace constants identical Tested: Focused PostgreSQL subscription mutation concurrency regression on JDK 25 Not-tested: Full suite because the change only replaces one fixed namespace key consistently --- .../dev/typetype/server/services/SubscriptionMutationLock.kt | 3 ++- .../dev/typetype/server/SubscriptionGroupsServiceTest.kt | 3 ++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/src/main/kotlin/dev/typetype/server/services/SubscriptionMutationLock.kt b/src/main/kotlin/dev/typetype/server/services/SubscriptionMutationLock.kt index 65caa657..8320b1ab 100644 --- a/src/main/kotlin/dev/typetype/server/services/SubscriptionMutationLock.kt +++ b/src/main/kotlin/dev/typetype/server/services/SubscriptionMutationLock.kt @@ -10,5 +10,6 @@ internal object SubscriptionMutationLock { ) } - private const val LOCK_NAMESPACE = 1_414_814_032 + // Precomputed PostgreSQL hashtext('subscriptions'). + private const val LOCK_NAMESPACE = 720_815_616 } diff --git a/src/test/kotlin/dev/typetype/server/SubscriptionGroupsServiceTest.kt b/src/test/kotlin/dev/typetype/server/SubscriptionGroupsServiceTest.kt index 74ba5762..f0bcfaca 100644 --- a/src/test/kotlin/dev/typetype/server/SubscriptionGroupsServiceTest.kt +++ b/src/test/kotlin/dev/typetype/server/SubscriptionGroupsServiceTest.kt @@ -206,4 +206,5 @@ class SubscriptionGroupsServiceTest { private fun subscriptionLockKey(userId: String): Int = userId.hashCode() and Int.MAX_VALUE } -private const val SUBSCRIPTION_LOCK_NAMESPACE = 1_414_814_032 +// Precomputed PostgreSQL hashtext('subscriptions'); must match SubscriptionMutationLock. +private const val SUBSCRIPTION_LOCK_NAMESPACE = 720_815_616