feat: add YouTube session backend

.env.example

@@ -4,5 +4,6 @@ DATABASE_PASSWORD=typetype
 
 DRAGONFLY_URL=redis://dragonfly:6379
 DOWNLOADER_SERVICE_URL=http://typetype-downloader:18093
+YOUTUBE_SESSION_ENCRYPTION_KEY=replace-with-at-least-32-random-characters
 
 ALLOWED_ORIGINS=http://localhost:5173

src/main/kotlin/dev/typetype/server/Application.kt

@@ -72,8 +72,10 @@ fun Application.module() {
     val cacheUrl = System.getenv("DRAGONFLY_URL") ?: "redis://localhost:6379"
     val subtitleServiceUrl = System.getenv("SUBTITLE_SERVICE_URL") ?: "http://typetype-token:8081"
     val downloaderServiceUrl = System.getenv("DOWNLOADER_SERVICE_URL") ?: "http://typetype-downloader:18093"
+    val youtubeSessionEncryptionKey = System.getenv("YOUTUBE_SESSION_ENCRYPTION_KEY")
+        ?: error("YOUTUBE_SESSION_ENCRYPTION_KEY is required")
     val cache = DragonflyService(cacheUrl)
-    val svc = ServiceRegistry(cache, subtitleServiceUrl)
+    val svc = ServiceRegistry(cache, subtitleServiceUrl, youtubeSessionEncryptionKey)
     val downloaderGatewayService = DownloaderGatewayService(downloaderServiceUrl)
     val openMojiProxyService = OpenMojiProxyService(cache)
     val internalHealthService = InternalHealthService(cache, downloaderGatewayService, subtitleServiceUrl)
@@ -82,7 +84,7 @@ fun Application.module() {
         internalObservabilityRoutes(internalHealthService::check)
         publicMetadataRoutes(instanceService::getInstance)
         rateLimit(STREAMS_ZONE) {
-            streamRoutes(svc.streamService)
+            streamRoutes(svc.streamService, authService, svc.youtubeSessionStreamService::getStreamInfo)
             manifestRoutes(svc.manifestService, svc.nativeManifestService, svc.hlsManifestService)
         }
         rateLimit(EXTRACTION_ZONE) {

src/main/kotlin/dev/typetype/server/ExtractionServiceRegistry.kt

@@ -0,0 +1,83 @@
+package dev.typetype.server
+
+import dev.typetype.server.cache.DragonflyService
+import dev.typetype.server.services.BilibiliRelatedService
+import dev.typetype.server.services.BilibiliTrendingService
+import dev.typetype.server.services.CachedChannelService
+import dev.typetype.server.services.CachedCommentService
+import dev.typetype.server.services.CachedManifestService
+import dev.typetype.server.services.CachedNativeManifestService
+import dev.typetype.server.services.CachedPodcastService
+import dev.typetype.server.services.CachedPublicPlaylistService
+import dev.typetype.server.services.CachedSearchService
+import dev.typetype.server.services.CachedStreamService
+import dev.typetype.server.services.CachedSuggestionService
+import dev.typetype.server.services.CachedTrendingService
+import dev.typetype.server.services.HlsManifestService
+import dev.typetype.server.services.ManifestService
+import dev.typetype.server.services.NativeManifestService
+import dev.typetype.server.services.NicoNicoTrendingService
+import dev.typetype.server.services.NicoVideoProxyService
+import dev.typetype.server.services.OkHttpProxyService
+import dev.typetype.server.services.PipePipeBulletCommentService
+import dev.typetype.server.services.PipePipeChannelService
+import dev.typetype.server.services.PipePipeCommentService
+import dev.typetype.server.services.PipePipePodcastService
+import dev.typetype.server.services.PipePipePublicPlaylistService
+import dev.typetype.server.services.PipePipeSearchService
+import dev.typetype.server.services.PipePipeStreamService
+import dev.typetype.server.services.PipePipeSuggestionService
+import dev.typetype.server.services.PipePipeTrendingService
+import dev.typetype.server.services.YouTubeSubtitleService
+import dev.typetype.server.services.YoutubeScopedChannelService
+import dev.typetype.server.services.YoutubeScopedCommentService
+import dev.typetype.server.services.YoutubeScopedPublicPlaylistService
+import dev.typetype.server.services.YoutubeScopedSearchService
+import dev.typetype.server.services.YoutubeScopedStreamService
+import dev.typetype.server.services.YoutubeScopedSuggestionService
+import dev.typetype.server.services.YoutubeScopedTrendingService
+import dev.typetype.server.services.YoutubeSessionCrypto
+import dev.typetype.server.services.YoutubeSessionService
+import dev.typetype.server.services.YoutubeSessionStreamService
+import okhttp3.OkHttpClient
+import java.util.concurrent.TimeUnit
+
+internal class ExtractionServiceRegistry(
+    cache: DragonflyService,
+    subtitleServiceUrl: String,
+    youtubeSessionEncryptionKey: String,
+) {
+    val httpClient = OkHttpClient()
+    val proxyHttpClient: OkHttpClient = httpClient.newBuilder()
+        .connectTimeout(10, TimeUnit.SECONDS)
+        .readTimeout(30, TimeUnit.SECONDS)
+        .followRedirects(true)
+        .build()
+    private val pipePipeStreamService = PipePipeStreamService(
+        cache,
+        YouTubeSubtitleService(httpClient, subtitleServiceUrl),
+        BilibiliRelatedService(),
+    )
+    val youtubeSessionService = YoutubeSessionService(YoutubeSessionCrypto.fromSecret(youtubeSessionEncryptionKey))
+    val youtubeSessionStreamService = YoutubeSessionStreamService(pipePipeStreamService, youtubeSessionService, cache)
+    val streamService = CachedStreamService(YoutubeScopedStreamService(pipePipeStreamService), cache)
+    val searchService = CachedSearchService(YoutubeScopedSearchService(PipePipeSearchService()), cache)
+    val trendingService = CachedTrendingService(
+        YoutubeScopedTrendingService(PipePipeTrendingService(BilibiliTrendingService(), NicoNicoTrendingService(httpClient))),
+        cache,
+    )
+    val commentService = CachedCommentService(YoutubeScopedCommentService(PipePipeCommentService()), cache)
+    val bulletCommentService = PipePipeBulletCommentService()
+    val channelService = CachedChannelService(YoutubeScopedChannelService(PipePipeChannelService()), cache)
+    val podcastService = CachedPodcastService(PipePipePodcastService(), cache)
+    val publicPlaylistService = CachedPublicPlaylistService(
+        YoutubeScopedPublicPlaylistService(PipePipePublicPlaylistService()),
+        cache,
+    )
+    val proxyService = OkHttpProxyService(proxyHttpClient)
+    val nicoVideoProxyService = NicoVideoProxyService()
+    val manifestService = CachedManifestService(ManifestService(streamService), cache)
+    val nativeManifestService = CachedNativeManifestService(NativeManifestService(), cache)
+    val hlsManifestService = HlsManifestService(streamService, proxyHttpClient)
+    val suggestionService = CachedSuggestionService(YoutubeScopedSuggestionService(PipePipeSuggestionService()), cache)
+}

src/main/kotlin/dev/typetype/server/ServiceRegistry.kt

@@ -1,38 +1,11 @@
 package dev.typetype.server
 import dev.typetype.server.cache.DragonflyService
-import dev.typetype.server.services.BilibiliRelatedService
-import dev.typetype.server.services.BilibiliTrendingService
 import dev.typetype.server.services.BlockedService
 import dev.typetype.server.services.BugReportService
-import dev.typetype.server.services.CachedChannelService
-import dev.typetype.server.services.CachedCommentService
-import dev.typetype.server.services.CachedManifestService
-import dev.typetype.server.services.CachedNativeManifestService
-import dev.typetype.server.services.CachedPodcastService
-import dev.typetype.server.services.CachedPublicPlaylistService
-import dev.typetype.server.services.CachedSearchService
-import dev.typetype.server.services.CachedStreamService
-import dev.typetype.server.services.CachedSuggestionService
-import dev.typetype.server.services.CachedTrendingService
 import dev.typetype.server.services.FavoritesService
 import dev.typetype.server.services.HistoryService
-import dev.typetype.server.services.HlsManifestService
 import dev.typetype.server.services.HomeRecommendationService
-import dev.typetype.server.services.ManifestService
-import dev.typetype.server.services.NativeManifestService
-import dev.typetype.server.services.NicoNicoTrendingService
-import dev.typetype.server.services.NicoVideoProxyService
 import dev.typetype.server.services.NotificationsService
-import dev.typetype.server.services.OkHttpProxyService
-import dev.typetype.server.services.PipePipeBulletCommentService
-import dev.typetype.server.services.PipePipeChannelService
-import dev.typetype.server.services.PipePipeCommentService
-import dev.typetype.server.services.PipePipePodcastService
-import dev.typetype.server.services.PipePipePublicPlaylistService
-import dev.typetype.server.services.PipePipeSearchService
-import dev.typetype.server.services.PipePipeStreamService
-import dev.typetype.server.services.PipePipeSuggestionService
-import dev.typetype.server.services.PipePipeTrendingService
 import dev.typetype.server.services.PlaylistService
 import dev.typetype.server.services.ProgressService
 import dev.typetype.server.services.SearchHistoryService
@@ -45,31 +18,33 @@ import dev.typetype.server.services.SubscriptionsService
 import dev.typetype.server.services.SubscriptionFeedCacheInvalidation
 import dev.typetype.server.services.SubscriptionFeedCacheInvalidator
 import dev.typetype.server.services.WatchLaterService
-import dev.typetype.server.services.YouTubeSubtitleService
 import dev.typetype.server.services.YoutubeTakeoutFactory
-import okhttp3.OkHttpClient
-import java.util.concurrent.TimeUnit
-internal class ServiceRegistry(cache: DragonflyService, subtitleServiceUrl: String) {
+internal class ServiceRegistry(
+    cache: DragonflyService,
+    subtitleServiceUrl: String,
+    youtubeSessionEncryptionKey: String,
+) {
     init {
         SubscriptionFeedCacheInvalidation.configure(SubscriptionFeedCacheInvalidator(cache))
     }
 
-    private val httpClient = OkHttpClient()
-    private val proxyHttpClient = httpClient.newBuilder().connectTimeout(10, TimeUnit.SECONDS).readTimeout(30, TimeUnit.SECONDS).followRedirects(true).build()
-    val streamService = CachedStreamService(PipePipeStreamService(cache, YouTubeSubtitleService(httpClient, subtitleServiceUrl), BilibiliRelatedService()), cache)
-    val searchService = CachedSearchService(PipePipeSearchService(), cache)
-    val trendingService = CachedTrendingService(PipePipeTrendingService(BilibiliTrendingService(), NicoNicoTrendingService(httpClient)), cache)
-    val commentService = CachedCommentService(PipePipeCommentService(), cache)
-    val bulletCommentService = PipePipeBulletCommentService()
-    val channelService = CachedChannelService(PipePipeChannelService(), cache)
-    val podcastService = CachedPodcastService(PipePipePodcastService(), cache)
-    val publicPlaylistService = CachedPublicPlaylistService(PipePipePublicPlaylistService(), cache)
-    val proxyService = OkHttpProxyService(proxyHttpClient)
-    val nicoVideoProxyService = NicoVideoProxyService()
-    val manifestService = CachedManifestService(ManifestService(streamService), cache)
-    val nativeManifestService = CachedNativeManifestService(NativeManifestService(), cache)
-    val hlsManifestService = HlsManifestService(streamService, proxyHttpClient)
-    val suggestionService = CachedSuggestionService(PipePipeSuggestionService(), cache)
+    private val extraction = ExtractionServiceRegistry(cache, subtitleServiceUrl, youtubeSessionEncryptionKey)
+    val youtubeSessionService = extraction.youtubeSessionService
+    val youtubeSessionStreamService = extraction.youtubeSessionStreamService
+    val streamService = extraction.streamService
+    val searchService = extraction.searchService
+    val trendingService = extraction.trendingService
+    val commentService = extraction.commentService
+    val bulletCommentService = extraction.bulletCommentService
+    val channelService = extraction.channelService
+    val podcastService = extraction.podcastService
+    val publicPlaylistService = extraction.publicPlaylistService
+    val proxyService = extraction.proxyService
+    val nicoVideoProxyService = extraction.nicoVideoProxyService
+    val manifestService = extraction.manifestService
+    val nativeManifestService = extraction.nativeManifestService
+    val hlsManifestService = extraction.hlsManifestService
+    val suggestionService = extraction.suggestionService
     val historyService = HistoryService()
     val subscriptionsService = SubscriptionsService()
     val subscriptionFeedService = SubscriptionFeedService(subscriptionsService, channelService, cache)

src/main/kotlin/dev/typetype/server/db/DatabaseFactory.kt

@@ -21,6 +21,8 @@ import dev.typetype.server.db.tables.PasswordResetTable
 import dev.typetype.server.db.tables.NotificationStatesTable
 import dev.typetype.server.db.tables.YoutubeTakeoutImportJobsTable
 import dev.typetype.server.db.tables.YoutubeTakeoutPlaylistKeysTable
+import dev.typetype.server.db.tables.YoutubeSessionPairingsTable
+import dev.typetype.server.db.tables.YoutubeSessionsTable
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.withContext
 import org.jetbrains.exposed.v1.jdbc.Database
@@ -59,6 +61,8 @@ object DatabaseFactory {
                 PasswordResetTable,
                 YoutubeTakeoutImportJobsTable,
                 YoutubeTakeoutPlaylistKeysTable,
+                YoutubeSessionsTable,
+                YoutubeSessionPairingsTable,
                 BugReportsTable,
                 NotificationStatesTable,
             )

src/main/kotlin/dev/typetype/server/db/tables/YoutubeSessionPairingsTable.kt

@@ -0,0 +1,11 @@
+package dev.typetype.server.db.tables
+
+import org.jetbrains.exposed.v1.core.Table
+
+object YoutubeSessionPairingsTable : Table("youtube_session_pairings") {
+    val code = text("code")
+    val userId = text("user_id")
+    val createdAt = long("created_at")
+    val expiresAt = long("expires_at")
+    override val primaryKey = PrimaryKey(code)
+}

src/main/kotlin/dev/typetype/server/db/tables/YoutubeSessionsTable.kt

@@ -0,0 +1,14 @@
+package dev.typetype.server.db.tables
+
+import org.jetbrains.exposed.v1.core.Table
+
+object YoutubeSessionsTable : Table("youtube_sessions") {
+    val userId = text("user_id")
+    val encryptedCookies = text("encrypted_cookies")
+    val encryptedPoToken = text("encrypted_po_token")
+    val status = text("status")
+    val createdAt = long("created_at")
+    val updatedAt = long("updated_at")
+    val lastUsedAt = long("last_used_at").default(0)
+    override val primaryKey = PrimaryKey(userId)
+}

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

@@ -0,0 +1,10 @@
+package dev.typetype.server.models
+
+import kotlinx.serialization.Serializable
+
+@Serializable
+data class YoutubeSessionCompleteRequest(
+    val code: String,
+    val cookies: String,
+    val poToken: String,
+)

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

@@ -0,0 +1,9 @@
+package dev.typetype.server.models
+
+import kotlinx.serialization.Serializable
+
+@Serializable
+data class YoutubeSessionPairingResponse(
+    val code: String,
+    val expiresAt: Long,
+)

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

@@ -0,0 +1,10 @@
+package dev.typetype.server.models
+
+import kotlinx.serialization.Serializable
+
+@Serializable
+data class YoutubeSessionStatusResponse(
+    val status: String,
+    val updatedAt: Long,
+    val lastUsedAt: Long,
+)

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

@@ -2,27 +2,85 @@ package dev.typetype.server.routes
 
 import dev.typetype.server.models.ErrorResponse
 import dev.typetype.server.models.ExtractionResult
+import dev.typetype.server.models.StreamResponse
+import dev.typetype.server.services.AuthService
 import dev.typetype.server.services.StreamService
+import dev.typetype.server.services.YOUTUBE_SESSION_RECONNECT_ERROR
 import io.ktor.http.HttpStatusCode
 import io.ktor.http.HttpHeaders
 import io.ktor.server.response.respond
 import io.ktor.server.routing.Route
 import io.ktor.server.routing.get
 
 private const val STREAMS_CACHE_CONTROL = "public, max-age=21600, stale-while-revalidate=3600"
+private const val AUTHENTICATED_STREAMS_CACHE_CONTROL = "no-store"
 
-fun Route.streamRoutes(streamService: StreamService) {
+fun Route.streamRoutes(
+    streamService: StreamService,
+    authService: AuthService? = null,
+    youtubeSessionStreamInfo: (suspend (String, String) -> ExtractionResult<StreamResponse>?)? = null,
+): Unit {
     get("/streams") {
         val url = call.request.queryParameters["url"]
             ?: return@get call.respond(HttpStatusCode.BadRequest, ErrorResponse("Missing 'url' parameter"))
 
-        when (val result = streamService.getStreamInfo(url)) {
+        val publicResult = streamService.getStreamInfo(url)
+        val userId = authService?.let { call.optionalJwtUserId(it) }
+        val sessionResult = if (
+            userId != null &&
+            youtubeSessionStreamInfo != null &&
+            publicResult.shouldTryYoutubeSession()
+        ) {
+            youtubeSessionStreamInfo(userId, url)
+        } else {
+            null
+        }
+        val usedYoutubeSession = sessionResult != null
+        when (val result = publicResult.resolveWith(sessionResult)) {
             is ExtractionResult.Success -> {
-                call.response.headers.append(HttpHeaders.CacheControl, STREAMS_CACHE_CONTROL)
+                call.response.headers.append(
+                    HttpHeaders.CacheControl,
+                    if (usedYoutubeSession) AUTHENTICATED_STREAMS_CACHE_CONTROL else STREAMS_CACHE_CONTROL,
+                )
                 call.respond(result.data)
             }
-            is ExtractionResult.BadRequest -> call.respond(HttpStatusCode.BadRequest, ErrorResponse(result.message))
+            is ExtractionResult.BadRequest -> call.respond(HttpStatusCode.BadRequest, result.toErrorResponse())
             is ExtractionResult.Failure -> call.respond(HttpStatusCode.UnprocessableEntity, ErrorResponse(result.message))
         }
     }
 }
+
+private fun ExtractionResult<StreamResponse>.shouldTryYoutubeSession(): Boolean = when (this) {
+    is ExtractionResult.Success -> data.hlsUrl.isBlank()
+    is ExtractionResult.BadRequest -> true
+    is ExtractionResult.Failure -> true
+}
+
+private fun ExtractionResult<StreamResponse>.resolveWith(
+    sessionResult: ExtractionResult<StreamResponse>?,
+): ExtractionResult<StreamResponse> {
+    if (sessionResult == null) return this
+    if (this is ExtractionResult.Success && sessionResult is ExtractionResult.Success) {
+        return ExtractionResult.Success(data.mergeWithSession(sessionResult.data))
+    }
+    if (this is ExtractionResult.Success) return this
+    return sessionResult
+}
+
+private fun StreamResponse.mergeWithSession(session: StreamResponse): StreamResponse {
+    val base = if (session.playableStreamCount() > playableStreamCount()) session else this
+    return base.copy(
+        hlsUrl = base.hlsUrl.ifBlank { session.hlsUrl.ifBlank { hlsUrl } },
+        dashMpdUrl = base.dashMpdUrl.ifBlank { session.dashMpdUrl.ifBlank { dashMpdUrl } },
+    )
+}
+
+private fun StreamResponse.playableStreamCount(): Int =
+    videoStreams.size + videoOnlyStreams.size + audioStreams.size
+
+private fun ExtractionResult.BadRequest.toErrorResponse(): ErrorResponse =
+    if (message == YOUTUBE_SESSION_RECONNECT_ERROR) {
+        ErrorResponse(message, "youtube_session_needs_reconnect")
+    } else {
+        ErrorResponse(message)
+    }

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

@@ -19,4 +19,11 @@ suspend fun ApplicationCall.withJwtAuth(authService: AuthService, block: suspend
         return
     }
     block(userId)
-}
+}
+
+fun ApplicationCall.optionalJwtUserId(authService: AuthService): String? {
+    val authHeader = request.headers["Authorization"]
+    if (authHeader == null || !authHeader.startsWith("Bearer ")) return null
+    val token = authHeader.substringAfter("Bearer ")
+    return authService.verify(token)
+}

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

@@ -28,6 +28,7 @@ internal fun Route.userDataRoutes(
     searchHistoryRoutes(svc.searchHistoryService, authService)
     blockedRoutes(svc.blockedService, authService)
     notificationsRoutes(svc.notificationsService, authService)
+    youtubeSessionRoutes(svc.youtubeSessionService, authService)
     youtubeTakeoutImportRoutes(svc.youtubeTakeoutImportService, authService)
     profileRoutes(profileService, avatarService, authService)
     bugReportRoutes(bugReportService, authService)