1 import * as debug from 'debug'
2 import { merge, Observable, of, ReplaySubject, Subject } from 'rxjs'
3 import { catchError, filter, map, switchMap, tap } from 'rxjs/operators'
4 import { HttpClient, HttpParams } from '@angular/common/http'
5 import { Injectable, NgZone } from '@angular/core'
6 import { ComponentPaginationLight, RestExtractor, RestService } from '@app/core'
7 import { buildBulkObservable } from '@app/helpers'
8 import { Video, VideoChannel, VideoChannelService, VideoService } from '@app/shared/shared-main'
9 import { ResultList, VideoChannel as VideoChannelServer, VideoSortField } from '@shared/models'
10 import { environment } from '../../../environments/environment'
12 const logger = debug('peertube:subscriptions:UserSubscriptionService')
14 type SubscriptionExistResult = { [ uri: string ]: boolean }
15 type SubscriptionExistResultObservable = { [ uri: string ]: Observable<boolean> }
18 export class UserSubscriptionService {
19 static BASE_USER_SUBSCRIPTIONS_URL = environment.apiUrl + '/api/v1/users/me/subscriptions'
21 // Use a replay subject because we "next" a value before subscribing
22 private existsSubject = new ReplaySubject<string>(1)
23 private readonly existsObservable: Observable<SubscriptionExistResult>
25 private myAccountSubscriptionCache: SubscriptionExistResult = {}
26 private myAccountSubscriptionCacheObservable: SubscriptionExistResultObservable = {}
27 private myAccountSubscriptionCacheSubject = new Subject<SubscriptionExistResult>()
30 private authHttp: HttpClient,
31 private restExtractor: RestExtractor,
32 private videoService: VideoService,
33 private restService: RestService,
34 private ngZone: NgZone
36 this.existsObservable = merge(
40 notifierObservable: this.existsSubject,
41 bulkGet: this.doSubscriptionsExist.bind(this)
44 this.myAccountSubscriptionCacheSubject
48 getUserSubscriptionVideos (parameters: {
49 videoPagination: ComponentPaginationLight
52 }): Observable<ResultList<Video>> {
53 const { videoPagination, sort, skipCount } = parameters
54 const pagination = this.restService.componentPaginationToRestPagination(videoPagination)
56 let params = new HttpParams()
57 params = this.restService.addRestGetParams(params, pagination, sort)
59 if (skipCount) params = params.set('skipCount', skipCount + '')
62 .get<ResultList<Video>>(UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL + '/videos', { params })
64 switchMap(res => this.videoService.extractVideos(res)),
65 catchError(err => this.restExtractor.handleError(err))
73 deleteSubscription (nameWithHost: string) {
74 const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL + '/' + nameWithHost
76 return this.authHttp.delete(url)
78 map(this.restExtractor.extractDataBool),
80 this.myAccountSubscriptionCache[nameWithHost] = false
82 this.myAccountSubscriptionCacheSubject.next(this.myAccountSubscriptionCache)
84 catchError(err => this.restExtractor.handleError(err))
88 addSubscription (nameWithHost: string) {
89 const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL
91 const body = { uri: nameWithHost }
92 return this.authHttp.post(url, body)
94 map(this.restExtractor.extractDataBool),
96 this.myAccountSubscriptionCache[nameWithHost] = true
98 this.myAccountSubscriptionCacheSubject.next(this.myAccountSubscriptionCache)
100 catchError(err => this.restExtractor.handleError(err))
104 listSubscriptions (parameters: {
105 pagination: ComponentPaginationLight
107 }): Observable<ResultList<VideoChannel>> {
108 const { pagination, search } = parameters
109 const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL
111 const restPagination = this.restService.componentPaginationToRestPagination(pagination)
113 let params = new HttpParams()
114 params = this.restService.addRestGetParams(params, restPagination)
115 if (search) params = params.append('search', search)
117 return this.authHttp.get<ResultList<VideoChannelServer>>(url, { params })
119 map(res => VideoChannelService.extractVideoChannels(res)),
120 catchError(err => this.restExtractor.handleError(err))
125 * SubscriptionExist part
128 listenToMyAccountSubscriptionCacheSubject () {
129 return this.myAccountSubscriptionCacheSubject.asObservable()
132 listenToSubscriptionCacheChange (nameWithHost: string) {
133 if (nameWithHost in this.myAccountSubscriptionCacheObservable) {
134 return this.myAccountSubscriptionCacheObservable[nameWithHost]
137 const obs = this.existsObservable
139 filter(existsResult => existsResult[nameWithHost] !== undefined),
140 map(existsResult => existsResult[nameWithHost])
143 this.myAccountSubscriptionCacheObservable[nameWithHost] = obs
147 doesSubscriptionExist (nameWithHost: string) {
148 logger('Running subscription check for %d.', nameWithHost)
150 if (nameWithHost in this.myAccountSubscriptionCache) {
151 logger('Found cache for %d.', nameWithHost)
153 return of(this.myAccountSubscriptionCache[nameWithHost])
156 this.existsSubject.next(nameWithHost)
158 logger('Fetching from network for %d.', nameWithHost)
159 return this.existsObservable.pipe(
160 filter(existsResult => existsResult[nameWithHost] !== undefined),
161 map(existsResult => existsResult[nameWithHost]),
162 tap(result => this.myAccountSubscriptionCache[nameWithHost] = result)
166 private doSubscriptionsExist (uris: string[]): Observable<SubscriptionExistResult> {
167 const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL + '/exist'
168 let params = new HttpParams()
170 params = this.restService.addObjectParams(params, { uris })
172 return this.authHttp.get<SubscriptionExistResult>(url, { params })
175 this.myAccountSubscriptionCache = {
176 ...this.myAccountSubscriptionCache,
180 catchError(err => this.restExtractor.handleError(err))