client/src/app/shared/shared-user-subscription/user-subscription.service.ts

   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 } 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 { ActorFollow, ResultList, VideoChannel as VideoChannelServer, VideoSortField } from '@shared/models'
  10 import { environment } from '../../../environments/environment'
  11
  12 const logger = debug('peertube:subscriptions:UserSubscriptionService')
  13
  14 type SubscriptionExistResult = { [ uri: string ]: boolean }
  15 type SubscriptionExistResultObservable = { [ uri: string ]: Observable<boolean> }
  16
  17 @Injectable()
  18 export class UserSubscriptionService {
  19   static BASE_USER_SUBSCRIPTIONS_URL = environment.apiUrl + '/api/v1/users/me/subscriptions'
  20   static BASE_VIDEO_CHANNELS_URL = environment.apiUrl + '/api/v1/video-channels'
  21   static BASE_ACCOUNTS_URL = environment.apiUrl + '/api/v1/accounts'
  22
  23   // Use a replay subject because we "next" a value before subscribing
  24   private existsSubject = new ReplaySubject<string>(1)
  25   private readonly existsObservable: Observable<SubscriptionExistResult>
  26
  27   private myAccountSubscriptionCache: SubscriptionExistResult = {}
  28   private myAccountSubscriptionCacheObservable: SubscriptionExistResultObservable = {}
  29   private myAccountSubscriptionCacheSubject = new Subject<SubscriptionExistResult>()
  30
  31   constructor (
  32     private authHttp: HttpClient,
  33     private restExtractor: RestExtractor,
  34     private videoService: VideoService,
  35     private restService: RestService
  36   ) {
  37     this.existsObservable = merge(
  38       buildBulkObservable({
  39         time: 500,
  40         notifierObservable: this.existsSubject,
  41         bulkGet: this.doSubscriptionsExist.bind(this)
  42       }).pipe(map(r => r.response)),
  43
  44       this.myAccountSubscriptionCacheSubject
  45     )
  46   }
  47
  48   listFollowers (parameters: {
  49     pagination: ComponentPaginationLight
  50     nameWithHost: string
  51     search?: string
  52   }) {
  53     const { pagination, nameWithHost, search } = parameters
  54
  55     let url = `${UserSubscriptionService.BASE_ACCOUNTS_URL}/${nameWithHost}/followers`
  56
  57     let params = new HttpParams()
  58     params = this.restService.addRestGetParams(params, this.restService.componentToRestPagination(pagination), '-createdAt')
  59
  60     if (search) {
  61       const filters = this.restService.parseQueryStringFilter(search, {
  62         channel: {
  63           prefix: 'channel:'
  64         }
  65       })
  66
  67       if (filters.channel) {
  68         url = `${UserSubscriptionService.BASE_VIDEO_CHANNELS_URL}/${filters.channel}/followers`
  69       }
  70
  71       params = this.restService.addObjectParams(params, { search: filters.search })
  72     }
  73
  74     return this.authHttp
  75       .get<ResultList<ActorFollow>>(url, { params })
  76       .pipe(
  77         catchError(err => this.restExtractor.handleError(err))
  78       )
  79   }
  80
  81   getUserSubscriptionVideos (parameters: {
  82     videoPagination: ComponentPaginationLight
  83     sort: VideoSortField
  84     skipCount?: boolean
  85   }): Observable<ResultList<Video>> {
  86     const { videoPagination, sort, skipCount } = parameters
  87     const pagination = this.restService.componentToRestPagination(videoPagination)
  88
  89     let params = new HttpParams()
  90     params = this.restService.addRestGetParams(params, pagination, sort)
  91
  92     if (skipCount) params = params.set('skipCount', skipCount + '')
  93
  94     return this.authHttp
  95                .get<ResultList<Video>>(UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL + '/videos', { params })
  96                .pipe(
  97                  switchMap(res => this.videoService.extractVideos(res)),
  98                  catchError(err => this.restExtractor.handleError(err))
  99                )
 100   }
 101
 102   /**
 103    * Subscription part
 104    */
 105
 106   deleteSubscription (nameWithHost: string) {
 107     const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL + '/' + nameWithHost
 108
 109     return this.authHttp.delete(url)
 110                .pipe(
 111                  map(this.restExtractor.extractDataBool),
 112                  tap(() => {
 113                    this.myAccountSubscriptionCache[nameWithHost] = false
 114
 115                    this.myAccountSubscriptionCacheSubject.next(this.myAccountSubscriptionCache)
 116                  }),
 117                  catchError(err => this.restExtractor.handleError(err))
 118                )
 119   }
 120
 121   addSubscription (nameWithHost: string) {
 122     const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL
 123
 124     const body = { uri: nameWithHost }
 125     return this.authHttp.post(url, body)
 126                .pipe(
 127                  map(this.restExtractor.extractDataBool),
 128                  tap(() => {
 129                    this.myAccountSubscriptionCache[nameWithHost] = true
 130
 131                    this.myAccountSubscriptionCacheSubject.next(this.myAccountSubscriptionCache)
 132                  }),
 133                  catchError(err => this.restExtractor.handleError(err))
 134                )
 135   }
 136
 137   listSubscriptions (parameters: {
 138     pagination: ComponentPaginationLight
 139     search: string
 140   }): Observable<ResultList<VideoChannel>> {
 141     const { pagination, search } = parameters
 142     const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL
 143
 144     const restPagination = this.restService.componentToRestPagination(pagination)
 145
 146     let params = new HttpParams()
 147     params = this.restService.addRestGetParams(params, restPagination)
 148     if (search) params = params.append('search', search)
 149
 150     return this.authHttp.get<ResultList<VideoChannelServer>>(url, { params })
 151                .pipe(
 152                  map(res => VideoChannelService.extractVideoChannels(res)),
 153                  catchError(err => this.restExtractor.handleError(err))
 154                )
 155   }
 156
 157   /**
 158    * SubscriptionExist part
 159    */
 160
 161   listenToMyAccountSubscriptionCacheSubject () {
 162     return this.myAccountSubscriptionCacheSubject.asObservable()
 163   }
 164
 165   listenToSubscriptionCacheChange (nameWithHost: string) {
 166     if (nameWithHost in this.myAccountSubscriptionCacheObservable) {
 167       return this.myAccountSubscriptionCacheObservable[nameWithHost]
 168     }
 169
 170     const obs = this.existsObservable
 171                     .pipe(
 172                       filter(existsResult => existsResult[nameWithHost] !== undefined),
 173                       map(existsResult => existsResult[nameWithHost])
 174                     )
 175
 176     this.myAccountSubscriptionCacheObservable[nameWithHost] = obs
 177     return obs
 178   }
 179
 180   doesSubscriptionExist (nameWithHost: string) {
 181     logger('Running subscription check for %d.', nameWithHost)
 182
 183     if (nameWithHost in this.myAccountSubscriptionCache) {
 184       logger('Found cache for %d.', nameWithHost)
 185
 186       return of(this.myAccountSubscriptionCache[nameWithHost])
 187     }
 188
 189     this.existsSubject.next(nameWithHost)
 190
 191     logger('Fetching from network for %d.', nameWithHost)
 192     return this.existsObservable.pipe(
 193       filter(existsResult => existsResult[nameWithHost] !== undefined),
 194       map(existsResult => existsResult[nameWithHost]),
 195       tap(result => this.myAccountSubscriptionCache[nameWithHost] = result)
 196     )
 197   }
 198
 199   private doSubscriptionsExist (uris: string[]): Observable<SubscriptionExistResult> {
 200     const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL + '/exist'
 201     let params = new HttpParams()
 202
 203     params = this.restService.addObjectParams(params, { uris })
 204
 205     return this.authHttp.get<SubscriptionExistResult>(url, { params })
 206                .pipe(
 207                  tap(res => {
 208                    this.myAccountSubscriptionCache = {
 209                      ...this.myAccountSubscriptionCache,
 210                      ...res
 211                    }
 212                  }),
 213                  catchError(err => this.restExtractor.handleError(err))
 214                )
 215   }
 216 }