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                  tap(() => {
 112                    this.myAccountSubscriptionCache[nameWithHost] = false
 113
 114                    this.myAccountSubscriptionCacheSubject.next(this.myAccountSubscriptionCache)
 115                  }),
 116                  catchError(err => this.restExtractor.handleError(err))
 117                )
 118   }
 119
 120   addSubscription (nameWithHost: string) {
 121     const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL
 122
 123     const body = { uri: nameWithHost }
 124     return this.authHttp.post(url, body)
 125                .pipe(
 126                  tap(() => {
 127                    this.myAccountSubscriptionCache[nameWithHost] = true
 128
 129                    this.myAccountSubscriptionCacheSubject.next(this.myAccountSubscriptionCache)
 130                  }),
 131                  catchError(err => this.restExtractor.handleError(err))
 132                )
 133   }
 134
 135   listSubscriptions (parameters: {
 136     pagination: ComponentPaginationLight
 137     search: string
 138   }): Observable<ResultList<VideoChannel>> {
 139     const { pagination, search } = parameters
 140     const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL
 141
 142     const restPagination = this.restService.componentToRestPagination(pagination)
 143
 144     let params = new HttpParams()
 145     params = this.restService.addRestGetParams(params, restPagination)
 146     if (search) params = params.append('search', search)
 147
 148     return this.authHttp.get<ResultList<VideoChannelServer>>(url, { params })
 149                .pipe(
 150                  map(res => VideoChannelService.extractVideoChannels(res)),
 151                  catchError(err => this.restExtractor.handleError(err))
 152                )
 153   }
 154
 155   /**
 156    * SubscriptionExist part
 157    */
 158
 159   listenToMyAccountSubscriptionCacheSubject () {
 160     return this.myAccountSubscriptionCacheSubject.asObservable()
 161   }
 162
 163   listenToSubscriptionCacheChange (nameWithHost: string) {
 164     if (nameWithHost in this.myAccountSubscriptionCacheObservable) {
 165       return this.myAccountSubscriptionCacheObservable[nameWithHost]
 166     }
 167
 168     const obs = this.existsObservable
 169                     .pipe(
 170                       filter(existsResult => existsResult[nameWithHost] !== undefined),
 171                       map(existsResult => existsResult[nameWithHost])
 172                     )
 173
 174     this.myAccountSubscriptionCacheObservable[nameWithHost] = obs
 175     return obs
 176   }
 177
 178   doesSubscriptionExist (nameWithHost: string) {
 179     logger('Running subscription check for %d.', nameWithHost)
 180
 181     if (nameWithHost in this.myAccountSubscriptionCache) {
 182       logger('Found cache for %d.', nameWithHost)
 183
 184       return of(this.myAccountSubscriptionCache[nameWithHost])
 185     }
 186
 187     this.existsSubject.next(nameWithHost)
 188
 189     logger('Fetching from network for %d.', nameWithHost)
 190     return this.existsObservable.pipe(
 191       filter(existsResult => existsResult[nameWithHost] !== undefined),
 192       map(existsResult => existsResult[nameWithHost]),
 193       tap(result => this.myAccountSubscriptionCache[nameWithHost] = result)
 194     )
 195   }
 196
 197   private doSubscriptionsExist (uris: string[]): Observable<SubscriptionExistResult> {
 198     const url = UserSubscriptionService.BASE_USER_SUBSCRIPTIONS_URL + '/exist'
 199     let params = new HttpParams()
 200
 201     params = this.restService.addObjectParams(params, { uris })
 202
 203     return this.authHttp.get<SubscriptionExistResult>(url, { params })
 204                .pipe(
 205                  tap(res => {
 206                    this.myAccountSubscriptionCache = {
 207                      ...this.myAccountSubscriptionCache,
 208                      ...res
 209                    }
 210                  }),
 211                  catchError(err => this.restExtractor.handleError(err))
 212                )
 213   }
 214 }