feat: expose and bulk-update subscription group memberships

openapi.yaml

@@ -45,6 +45,7 @@ paths:
   /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/group-memberships: { $ref: ./openapi/paths/subscriptions.yaml#/SubscriptionGroupMemberships }
   /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 }
@@ -160,9 +161,11 @@ components:
     SavedPlaylistItem: { $ref: ./openapi/components/media.yaml#/SavedPlaylistItem }
     SavedPlaylistRequest: { $ref: ./openapi/components/media.yaml#/SavedPlaylistRequest }
     SubscriptionItem: { $ref: ./openapi/components/subscriptions.yaml#/SubscriptionItem }
+    SubscriptionGroupMembershipItem: { $ref: ./openapi/components/subscriptions.yaml#/SubscriptionGroupMembershipItem }
     SubscriptionGroupItem: { $ref: ./openapi/components/subscriptions.yaml#/SubscriptionGroupItem }
     SubscriptionGroupRequest: { $ref: ./openapi/components/subscriptions.yaml#/SubscriptionGroupRequest }
     SubscriptionGroupMembershipRequest: { $ref: ./openapi/components/subscriptions.yaml#/SubscriptionGroupMembershipRequest }
+    SubscriptionGroupMembershipBatchRequest: { $ref: ./openapi/components/subscriptions.yaml#/SubscriptionGroupMembershipBatchRequest }
     SubscriptionFeedResponse: { $ref: ./openapi/components/subscriptions.yaml#/SubscriptionFeedResponse }
     SubscriptionFeedPreparingResponse: { $ref: ./openapi/components/subscriptions.yaml#/SubscriptionFeedPreparingResponse }
     RssFeedRequest: { $ref: ./openapi/components/rss.yaml#/RssFeedRequest }

openapi/components/subscriptions.yaml

@@ -6,6 +6,18 @@ SubscriptionItem:
     name: { type: string }
     avatarUrl: { type: string }
     subscribedAt: { type: integer, format: int64 }
+SubscriptionGroupMembershipItem:
+  type: object
+  required: [channelUrl, name, avatarUrl, subscribedAt, groupIds]
+  properties:
+    channelUrl: { type: string, minLength: 1 }
+    name: { type: string }
+    avatarUrl: { type: string }
+    subscribedAt: { type: integer, format: int64 }
+    groupIds:
+      type: array
+      uniqueItems: true
+      items: { type: string, format: uuid }
 SubscriptionCreateRequest:
   type: object
   required: [channelUrl, name, avatarUrl]
@@ -31,7 +43,16 @@ SubscriptionGroupMembershipRequest:
   type: object
   required: [channelUrl]
   properties:
-    channelUrl: { type: string, minLength: 1 }
+    channelUrl: { type: string, minLength: 1, maxLength: 2048 }
+SubscriptionGroupMembershipBatchRequest:
+  type: object
+  required: [channelUrls]
+  properties:
+    channelUrls:
+      type: array
+      minItems: 1
+      maxItems: 500
+      items: { type: string, minLength: 1, maxLength: 2048 }
 SubscriptionFeedResponse:
   type: object
   required: [videos, nextpage, generation, generatedAt, refreshing]

openapi/paths/subscriptions.yaml

@@ -52,6 +52,20 @@ Subscriptions:
       '400': { $ref: ../components/common.yaml#/JsonError }
       '401': { $ref: ../components/common.yaml#/JsonError }
       '404': { $ref: ../components/common.yaml#/JsonError }
+SubscriptionGroupMemberships:
+  get:
+    tags: [user-data]
+    summary: List subscriptions with their group memberships
+    description: Returns every current subscription once, with all account-owned group IDs assigned to that channel. Ungrouped subscriptions have an empty groupIds array.
+    responses:
+      '200':
+        description: Account-scoped subscriptions and their complete group memberships.
+        content:
+          application/json:
+            schema:
+              type: array
+              items: { $ref: ../components/subscriptions.yaml#/SubscriptionGroupMembershipItem }
+      '401': { $ref: ../components/common.yaml#/JsonError }
 SubscriptionGroups:
   get:
     tags: [user-data]
@@ -117,29 +131,44 @@ SubscriptionGroupChannels:
       schema: { type: string, format: uuid }
   put:
     tags: [user-data]
-    summary: Add a subscribed channel to a group
+    summary: Add subscribed channels to a group
+    description: Accepts the original singular channelUrl request or up to 500 channelUrls. Batch additions are atomic, canonicalized and deduplicated. Request bodies are limited to 1 MiB.
     requestBody:
       required: true
       content:
         application/json:
-          schema: { $ref: ../components/subscriptions.yaml#/SubscriptionGroupMembershipRequest }
+          schema:
+            oneOf:
+              - { $ref: ../components/subscriptions.yaml#/SubscriptionGroupMembershipRequest }
+              - { $ref: ../components/subscriptions.yaml#/SubscriptionGroupMembershipBatchRequest }
     responses:
       '204': { description: Membership exists. }
       '400': { $ref: ../components/common.yaml#/JsonError }
       '401': { $ref: ../components/common.yaml#/JsonError }
+      '413': { $ref: ../components/common.yaml#/JsonError }
       '404': { $ref: ../components/common.yaml#/JsonError }
   delete:
     tags: [user-data]
-    summary: Remove a subscribed channel from a group
+    summary: Remove subscribed channels from a group
+    description: The url query parameter preserves the original singular contract. A JSON request body can remove one channel or up to 500 channelUrls atomically; absent batch memberships are ignored. Supplying both url and any non-empty request body returns 400. Request bodies are limited to 1 MiB.
     parameters:
       - name: url
         in: query
-        required: true
-        schema: { type: string, minLength: 1 }
+        required: false
+        schema: { type: string, minLength: 1, maxLength: 2048 }
+    requestBody:
+      required: false
+      content:
+        application/json:
+          schema:
+            oneOf:
+              - { $ref: ../components/subscriptions.yaml#/SubscriptionGroupMembershipRequest }
+              - { $ref: ../components/subscriptions.yaml#/SubscriptionGroupMembershipBatchRequest }
     responses:
       '204': { description: Membership deleted. }
       '400': { $ref: ../components/common.yaml#/JsonError }
       '401': { $ref: ../components/common.yaml#/JsonError }
+      '413': { $ref: ../components/common.yaml#/JsonError }
       '404': { $ref: ../components/common.yaml#/JsonError }
 SubscriptionFeed:
   get:

src/main/kotlin/dev/typetype/server/models/SubscriptionGroupMembershipItem.kt

@@ -0,0 +1,12 @@
+package dev.typetype.server.models
+
+import kotlinx.serialization.Serializable
+
+@Serializable
+data class SubscriptionGroupMembershipItem(
+    val channelUrl: String,
+    val name: String,
+    val avatarUrl: String,
+    val subscribedAt: Long,
+    val groupIds: List<String>,
+)

src/main/kotlin/dev/typetype/server/models/SubscriptionGroupMembershipRequest.kt

@@ -3,4 +3,7 @@ package dev.typetype.server.models
 import kotlinx.serialization.Serializable
 
 @Serializable
-data class SubscriptionGroupMembershipRequest(val channelUrl: String)
+data class SubscriptionGroupMembershipRequest(
+    val channelUrl: String? = null,
+    val channelUrls: List<String>? = null,
+)

src/main/kotlin/dev/typetype/server/routes/SubscriptionGroupMembershipRequestBody.kt

@@ -0,0 +1,85 @@
+package dev.typetype.server.routes
+
+import dev.typetype.server.models.ErrorResponse
+import dev.typetype.server.models.SubscriptionGroupMembershipRequest
+import dev.typetype.server.services.SubscriptionGroupsService
+import io.ktor.http.ContentType
+import io.ktor.http.HttpHeaders
+import io.ktor.http.HttpStatusCode
+import io.ktor.server.application.ApplicationCall
+import io.ktor.server.request.contentType
+import io.ktor.server.request.receiveChannel
+import io.ktor.server.response.respond
+import io.ktor.utils.io.ByteReadChannel
+import io.ktor.utils.io.jvm.javaio.toInputStream
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.withContext
+import kotlinx.serialization.SerializationException
+import kotlinx.serialization.json.Json
+import java.io.ByteArrayOutputStream
+
+private const val MAX_MEMBERSHIP_REQUEST_BODY_BYTES = 1024 * 1024
+private const val MAX_MEMBERSHIP_CHANNEL_URL_LENGTH = 2048
+private val membershipRequestJson = Json { ignoreUnknownKeys = true }
+
+internal sealed interface MembershipChannels {
+    data class Single(val channelUrl: String) : MembershipChannels
+    data class Batch(val channelUrls: List<String>) : MembershipChannels
+}
+
+internal suspend fun ApplicationCall.receiveMembershipChannels(body: ByteArray): MembershipChannels? {
+    val request = if (request.contentType().match(ContentType.Application.Json)) {
+        try {
+            membershipRequestJson.decodeFromString<SubscriptionGroupMembershipRequest>(body.decodeToString())
+        } catch (_: SerializationException) {
+            null
+        }
+    } else {
+        null
+    }
+    val channelUrl = request?.channelUrl?.takeIf(String::isValidMembershipChannelUrl)
+    val channelUrls = request?.channelUrls
+    val parsed = when {
+        channelUrl != null && channelUrls == null -> MembershipChannels.Single(channelUrl)
+        request?.channelUrl == null && channelUrls != null &&
+            channelUrls.size in 1..SubscriptionGroupsService.MAX_MEMBERSHIP_CHANNELS &&
+            channelUrls.all(String::isValidMembershipChannelUrl) -> MembershipChannels.Batch(channelUrls)
+        else -> null
+    }
+    if (parsed == null) respond(HttpStatusCode.BadRequest, ErrorResponse("Invalid request body"))
+    return parsed
+}
+
+internal suspend fun ApplicationCall.receiveMembershipBody(): ByteArray? {
+    val contentLength = request.headers[HttpHeaders.ContentLength]?.toLongOrNull()
+    if (contentLength != null && contentLength > MAX_MEMBERSHIP_REQUEST_BODY_BYTES) {
+        respondMembershipBodyTooLarge()
+        return null
+    }
+    val body = receiveChannel().readUpTo(MAX_MEMBERSHIP_REQUEST_BODY_BYTES)
+    if (body == null) respondMembershipBodyTooLarge()
+    return body
+}
+
+private suspend fun ByteReadChannel.readUpTo(maxBytes: Int): ByteArray? = withContext(Dispatchers.IO) {
+    toInputStream().use { input ->
+        ByteArrayOutputStream().use { output ->
+            val buffer = ByteArray(DEFAULT_BUFFER_SIZE)
+            while (true) {
+                val read = input.read(buffer)
+                if (read <= 0) break
+                if (output.size() + read > maxBytes) return@withContext null
+                output.write(buffer, 0, read)
+            }
+            output.toByteArray()
+        }
+    }
+}
+
+internal fun String.isValidMembershipChannelUrl(): Boolean =
+    isNotBlank() && length <= MAX_MEMBERSHIP_CHANNEL_URL_LENGTH
+
+private suspend fun ApplicationCall.respondMembershipBodyTooLarge() = respond(
+    HttpStatusCode.PayloadTooLarge,
+    ErrorResponse("Request body exceeds 1 MiB", "request_body_too_large"),
+)

src/main/kotlin/dev/typetype/server/routes/SubscriptionGroupsRoutes.kt

@@ -1,7 +1,6 @@
 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
@@ -45,21 +44,44 @@ fun Route.subscriptionGroupsRoutes(groupsService: SubscriptionGroupsService, aut
     put("/subscriptions/groups/{groupId}/channels") {
         call.withJwtAuth(authService) { userId ->
             val groupId = call.groupId() ?: return@withJwtAuth call.respondMissingGroupId()
-            val request = runCatching { call.receive<SubscriptionGroupMembershipRequest>() }.getOrElse {
-                return@withJwtAuth call.respond(HttpStatusCode.BadRequest, ErrorResponse("Invalid request body"))
+            val body = call.receiveMembershipBody() ?: return@withJwtAuth
+            when (val request = call.receiveMembershipChannels(body) ?: return@withJwtAuth) {
+                is MembershipChannels.Single -> call.respondMembership(
+                    groupsService.addSubscription(userId, groupId, request.channelUrl),
+                )
+                is MembershipChannels.Batch -> call.respondMembership(
+                    groupsService.addSubscriptions(userId, groupId, request.channelUrls),
+                )
             }
-            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))
+            val queryChannelUrl = call.request.queryParameters["url"]
+            if (queryChannelUrl != null && !queryChannelUrl.isValidMembershipChannelUrl()) {
+                return@withJwtAuth call.respond(HttpStatusCode.BadRequest, ErrorResponse("Invalid channel URL"))
+            }
+            val body = call.receiveMembershipBody() ?: return@withJwtAuth
+            if (queryChannelUrl != null) {
+                if (body.isNotEmpty()) {
+                    return@withJwtAuth call.respond(
+                        HttpStatusCode.BadRequest,
+                        ErrorResponse("Specify either url or a request body, not both"),
+                    )
+                }
+                return@withJwtAuth call.respondMembership(
+                    groupsService.removeSubscription(userId, groupId, queryChannelUrl),
+                )
+            }
+            when (val request = call.receiveMembershipChannels(body) ?: return@withJwtAuth) {
+                is MembershipChannels.Single -> call.respondMembership(
+                    groupsService.removeSubscription(userId, groupId, request.channelUrl),
+                )
+                is MembershipChannels.Batch -> call.respondMembership(
+                    groupsService.removeSubscriptions(userId, groupId, request.channelUrls),
+                )
+            }
         }
     }
 }

src/main/kotlin/dev/typetype/server/routes/SubscriptionsRoutes.kt

@@ -27,6 +27,11 @@ fun Route.subscriptionsRoutes(
     warmupService: HomeRecommendationWarmup = NoopHomeRecommendationWarmup,
     groupsService: SubscriptionGroupsService = SubscriptionGroupsService(),
 ) {
+    get("/subscriptions/group-memberships") {
+        call.withJwtAuth(authService) { userId ->
+            call.respond(subscriptionsService.getAllWithGroupMemberships(userId))
+        }
+    }
     get("/subscriptions") {
         call.withJwtAuth(authService) { userId ->
             val parsed = call.parseSubscriptionSelection()

src/main/kotlin/dev/typetype/server/services/SubscriptionGroupsService.kt

@@ -9,6 +9,8 @@ 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.core.inList
+import org.jetbrains.exposed.v1.jdbc.batchInsert
 import org.jetbrains.exposed.v1.jdbc.deleteWhere
 import org.jetbrains.exposed.v1.jdbc.insertIgnore
 import org.jetbrains.exposed.v1.jdbc.selectAll
@@ -103,19 +105,31 @@ class SubscriptionGroupsService {
         userId: String,
         groupId: String,
         rawChannelUrl: String,
+    ): SubscriptionGroupMembershipResult = addSubscriptions(userId, groupId, listOf(rawChannelUrl))
+
+    suspend fun addSubscriptions(
+        userId: String,
+        groupId: String,
+        rawChannelUrls: List<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 {
-            (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()
+        val channelUrls = rawChannelUrls.mapTo(linkedSetOf(), ChannelUrlCanonicalizer::canonicalize)
+        val subscribed = SubscriptionsTable.selectAll().where {
+            (SubscriptionsTable.userId eq userId) and (SubscriptionsTable.channelUrl inList channelUrls)
+        }.mapTo(hashSetOf()) { it[SubscriptionsTable.channelUrl] }
+        if (subscribed.size != channelUrls.size) return@query SubscriptionGroupMembershipResult.SubscriptionNotFound
+        val existing = SubscriptionGroupMembershipsTable.selectAll().where {
+            (SubscriptionGroupMembershipsTable.groupId eq groupId) and
+                (SubscriptionGroupMembershipsTable.userId eq userId) and
+                (SubscriptionGroupMembershipsTable.channelUrl inList channelUrls)
+        }.mapTo(hashSetOf()) { it[SubscriptionGroupMembershipsTable.channelUrl] }
+        val addedAt = System.currentTimeMillis()
+        SubscriptionGroupMembershipsTable.batchInsert(channelUrls - existing, shouldReturnGeneratedValues = false) { url ->
+            this[SubscriptionGroupMembershipsTable.groupId] = groupId
+            this[SubscriptionGroupMembershipsTable.userId] = userId
+            this[SubscriptionGroupMembershipsTable.channelUrl] = url
+            this[SubscriptionGroupMembershipsTable.addedAt] = addedAt
         }
         SubscriptionGroupMembershipResult.Success
     }
@@ -138,6 +152,22 @@ class SubscriptionGroupsService {
         }
     }
 
+    suspend fun removeSubscriptions(
+        userId: String,
+        groupId: String,
+        rawChannelUrls: List<String>,
+    ): SubscriptionGroupMembershipResult = DatabaseFactory.query {
+        SubscriptionMutationLock.acquire(userId)
+        if (!groupExists(userId, groupId)) return@query SubscriptionGroupMembershipResult.GroupNotFound
+        val channelUrls = rawChannelUrls.mapTo(hashSetOf(), ChannelUrlCanonicalizer::canonicalize)
+        SubscriptionGroupMembershipsTable.deleteWhere {
+            (SubscriptionGroupMembershipsTable.groupId eq groupId) and
+                (SubscriptionGroupMembershipsTable.userId eq userId) and
+                (SubscriptionGroupMembershipsTable.channelUrl inList channelUrls)
+        }
+        SubscriptionGroupMembershipResult.Success
+    }
+
     suspend fun getChannelUrls(userId: String, groupId: String): List<String> = DatabaseFactory.query {
         SubscriptionGroupMembershipsTable.selectAll().where {
             (SubscriptionGroupMembershipsTable.groupId eq groupId) and
@@ -182,6 +212,7 @@ class SubscriptionGroupsService {
 
     companion object {
         const val MAX_GROUP_NAME_LENGTH = 100
+        const val MAX_MEMBERSHIP_CHANNELS = 500
         private const val UNIQUE_VIOLATION_SQL_STATE = "23505"
     }
 }