| 1 | import type { |
| 2 | AnalyticsAccountInput, |
| 3 | AnalyticsWorkInput, |
| 4 | ChannelAccountDataSnapshotPayload, |
| 5 | ChannelAccountMetricsSnapshot, |
| 6 | ChannelWorkAnalyticsResult, |
| 7 | ChannelWorkDataSnapshotPayload, |
| 8 | } from '../platforms/platforms.interface' |
| 9 | import { Injectable, Logger } from '@nestjs/common' |
| 10 | import { ChannelAccountDataSnapshotRepository, ChannelWorkDataSnapshotRepository } from '@yikart/channel-db' |
| 11 | import { AccountType, AppException, ResponseCode } from '@yikart/common' |
| 12 | import { AccountRepository } from '@yikart/mongodb' |
| 13 | import { AuthService } from '../auth/auth.service' |
| 14 | import { ChannelPlatformException, PlatformErrorCategory } from '../platforms/platforms.exception' |
| 15 | import { PlatformIntegrationRegistry } from '../platforms/platforms.registry' |
| 16 | import { RelayAccountException } from '../relay/relay-account.exception' |
| 17 | |
| 18 | @Injectable() |
| 19 | export class AnalyticsService { |
| 20 | private readonly logger = new Logger(AnalyticsService.name) |
| 21 | |
| 22 | constructor( |
| 23 | private readonly registry: PlatformIntegrationRegistry, |
| 24 | private readonly authService: AuthService, |
| 25 | private readonly accountRepository: AccountRepository, |
| 26 | private readonly accountSnapshotRepository: ChannelAccountDataSnapshotRepository, |
| 27 | private readonly workSnapshotRepository: ChannelWorkDataSnapshotRepository, |
| 28 | ) {} |
| 29 | |
| 30 | async fetchAccountAnalytics( |
| 31 | userId: string, |
| 32 | accountId: string, |
| 33 | params?: { since?: Date, until?: Date }, |
| 34 | ) { |
| 35 | const account = await this.accountRepository.getByIdAndUserId(accountId, userId) |
| 36 | if (!account) { |
| 37 | throw new AppException(ResponseCode.AccountNotFound) |
| 38 | } |
| 39 | const platform = account.type |
| 40 | if (account.relayAccountRef) { |
| 41 | throw new RelayAccountException(account.relayAccountRef, accountId) |
| 42 | } |
| 43 | const provider = this.registry.getAnalytics(platform) |
| 44 | if (!provider) { |
| 45 | return { platform, accountId, snapshots: [], message: 'Analytics not supported' } |
| 46 | } |
| 47 | |
| 48 | const credential = await this.authService.getValidCredential(accountId, userId) |
| 49 | |
| 50 | const input: AnalyticsAccountInput = { |
| 51 | accountId, |
| 52 | platform, |
| 53 | credential: { |
| 54 | accessToken: credential.accessToken, |
| 55 | refreshToken: credential.refreshToken, |
| 56 | expiresAt: credential.expiresAt, |
| 57 | scope: credential.scope, |
| 58 | platformUid: account.uid, |
| 59 | account: account.account, |
| 60 | }, |
| 61 | since: params?.since, |
| 62 | until: params?.until, |
| 63 | } |
| 64 | |
| 65 | let result: Awaited<ReturnType<typeof provider.fetchAccountAnalytics>> |
| 66 | try { |
| 67 | result = await provider.fetchAccountAnalytics(input) |
| 68 | } |
| 69 | catch (error) { |
| 70 | if (!( |
| 71 | platform === AccountType.YouTube |
| 72 | && error instanceof ChannelPlatformException |
| 73 | && error.category === PlatformErrorCategory.Auth |
| 74 | )) { |
| 75 | await this.authService.markAccountOfflineForCredentialFailure(accountId, error, 'platform_auth_failed') |
| 76 | throw error |
| 77 | } |
| 78 | |
| 79 | const refreshedCredential = await this.authService.refreshCredential(accountId, userId) |
| 80 | try { |
| 81 | result = await provider.fetchAccountAnalytics({ |
| 82 | ...input, |
| 83 | credential: { |
| 84 | ...input.credential, |
| 85 | accessToken: refreshedCredential.accessToken, |
| 86 | refreshToken: refreshedCredential.refreshToken, |
| 87 | expiresAt: refreshedCredential.expiresAt, |
| 88 | scope: refreshedCredential.scope, |
| 89 | }, |
| 90 | }) |
| 91 | } |
| 92 | catch (retryError) { |
| 93 | await this.authService.markAccountOfflineForCredentialFailure(accountId, retryError, 'platform_auth_failed') |
| 94 | throw retryError |
| 95 | } |
| 96 | } |
| 97 | const savedSnapshots = await this.saveAccountSnapshots( |
| 98 | userId, |
| 99 | platform, |
| 100 | accountId, |
| 101 | result.snapshots, |
| 102 | result.rawResponse, |
| 103 | ) |
| 104 | const latestSnapshot = savedSnapshots[savedSnapshots.length - 1] |
| 105 | |
| 106 | if (latestSnapshot?.metrics) { |
| 107 | await this.syncAccountMetrics(accountId, latestSnapshot.metrics, latestSnapshot.fetchedAt) |
| 108 | } |
| 109 | |
| 110 | return { |
| 111 | platform, |
| 112 | accountId, |
| 113 | profile: latestSnapshot?.profile ?? result.profile, |
| 114 | metrics: latestSnapshot?.metrics ?? result.metrics, |
| 115 | snapshots: savedSnapshots, |
| 116 | extra: latestSnapshot?.extra ?? result.extra, |
| 117 | snapshotId: latestSnapshot?.id, |
| 118 | fetchedAt: latestSnapshot?.fetchedAt, |
| 119 | } |
| 120 | } |
| 121 | |
| 122 | async fetchWorkAnalytics( |
| 123 | userId: string, |
| 124 | platform: AccountType, |
| 125 | platformWorkId: string, |
| 126 | accountId?: string, |
| 127 | params?: { since?: Date, until?: Date }, |
| 128 | ) { |
| 129 | if (!accountId) { |
| 130 | return { platform, accountId, platformWorkId, snapshots: [], message: 'Account required' } |
| 131 | } |
| 132 | |
| 133 | const account = await this.getPlatformAccount(userId, accountId, platform) |
| 134 | const provider = this.registry.getAnalytics(platform) |
| 135 | let platformResult: ChannelWorkAnalyticsResult | undefined |
| 136 | let platformError: unknown |
| 137 | |
| 138 | if (provider?.fetchWorkAnalytics) { |
| 139 | try { |
| 140 | const credential = await this.getCredentialContext(accountId, userId, account) |
| 141 | const input: AnalyticsWorkInput = { |
| 142 | accountId, |
| 143 | platformWorkId, |
| 144 | platform, |
| 145 | credential, |
| 146 | since: params?.since, |
| 147 | until: params?.until, |
| 148 | } |
| 149 | platformResult = await this.callPlatformProvider(accountId, () => provider.fetchWorkAnalytics!(input)) |
| 150 | if (this.hasWorkAnalyticsData(platformResult)) { |
| 151 | return this.toWorkAnalyticsResponse( |
| 152 | userId, |
| 153 | platform, |
| 154 | platformWorkId, |
| 155 | accountId, |
| 156 | platformResult, |
| 157 | ) |
| 158 | } |
| 159 | } |
| 160 | catch (error) { |
| 161 | platformError = error |
| 162 | this.logger.error( |
| 163 | error, |
| 164 | `Fetch channel work analytics from platform failed: platform=${platform}, accountId=${accountId}, platformWorkId=${platformWorkId}`, |
| 165 | ) |
| 166 | } |
| 167 | } |
| 168 | |
| 169 | if (platformError) { |
| 170 | throw platformError |
| 171 | } |
| 172 | if (platformResult) { |
| 173 | return this.toWorkAnalyticsResponse(userId, platform, platformWorkId, accountId, platformResult) |
| 174 | } |
| 175 | |
| 176 | return { platform, accountId, platformWorkId, snapshots: [], message: 'Work analytics not supported' } |
| 177 | } |
| 178 | |
| 179 | private async toWorkAnalyticsResponse( |
| 180 | userId: string, |
| 181 | platform: AccountType, |
| 182 | platformWorkId: string, |
| 183 | accountId: string | undefined, |
| 184 | result: ChannelWorkAnalyticsResult, |
| 185 | ) { |
| 186 | const savedSnapshots = await this.saveWorkSnapshots( |
| 187 | userId, |
| 188 | platform, |
| 189 | platformWorkId, |
| 190 | accountId, |
| 191 | result.snapshots, |
| 192 | result.rawResponse, |
| 193 | ) |
| 194 | const latestSnapshot = savedSnapshots[savedSnapshots.length - 1] |
| 195 | |
| 196 | return { |
| 197 | platform, |
| 198 | accountId, |
| 199 | platformWorkId, |
| 200 | work: latestSnapshot?.work ?? result.work, |
| 201 | metrics: latestSnapshot?.metrics ?? result.metrics, |
| 202 | snapshots: savedSnapshots, |
| 203 | extra: latestSnapshot?.extra ?? result.extra, |
| 204 | snapshotId: latestSnapshot?.id, |
| 205 | fetchedAt: latestSnapshot?.fetchedAt, |
| 206 | } |
| 207 | } |
| 208 | |
| 209 | private hasWorkAnalyticsData(result: ChannelWorkAnalyticsResult) { |
| 210 | return result.snapshots.length > 0 || Boolean(result.metrics) |
| 211 | } |
| 212 | |
| 213 | private async getPlatformAccount(userId: string, accountId: string, platform: AccountType) { |
| 214 | const account = await this.accountRepository.getByIdAndUserId(accountId, userId) |
| 215 | if (!account) { |
| 216 | throw new AppException(ResponseCode.AccountNotFound) |
| 217 | } |
| 218 | if (account.type !== platform) { |
| 219 | throw new AppException(ResponseCode.ChannelAuthPlatformMismatch) |
| 220 | } |
| 221 | if (account.relayAccountRef) { |
| 222 | throw new RelayAccountException(account.relayAccountRef, accountId) |
| 223 | } |
| 224 | return account |
| 225 | } |
| 226 | |
| 227 | private async getCredentialContext( |
| 228 | accountId: string, |
| 229 | userId: string, |
| 230 | account: { uid?: string, account?: string }, |
| 231 | ) { |
| 232 | const credential = await this.authService.getValidCredential(accountId, userId) |
| 233 | return { |
| 234 | accessToken: credential.accessToken, |
| 235 | refreshToken: credential.refreshToken, |
| 236 | expiresAt: credential.expiresAt, |
| 237 | scope: credential.scope, |
| 238 | platformUid: account.uid, |
| 239 | account: account.account, |
| 240 | } |
| 241 | } |
| 242 | |
| 243 | private async callPlatformProvider<T>(accountId: string, action: () => Promise<T>): Promise<T> { |
| 244 | try { |
| 245 | return await action() |
| 246 | } |
| 247 | catch (error) { |
| 248 | await this.authService.markAccountOfflineForCredentialFailure(accountId, error, 'platform_auth_failed') |
| 249 | throw error |
| 250 | } |
| 251 | } |
| 252 | |
| 253 | private async saveAccountSnapshots( |
| 254 | userId: string, |
| 255 | platform: AccountType, |
| 256 | accountId: string, |
| 257 | snapshots: ChannelAccountDataSnapshotPayload[], |
| 258 | rawResponse?: unknown, |
| 259 | ) { |
| 260 | if (snapshots.length === 0) { |
| 261 | return [] |
| 262 | } |
| 263 | return this.accountSnapshotRepository.createMany( |
| 264 | snapshots.map(snapshot => ({ |
| 265 | userId, |
| 266 | platform, |
| 267 | accountId, |
| 268 | platformUid: snapshot.platformUid, |
| 269 | snapshotAt: snapshot.snapshotAt, |
| 270 | fetchedAt: snapshot.fetchedAt ?? new Date(), |
| 271 | periodStartAt: snapshot.periodStartAt, |
| 272 | periodEndAt: snapshot.periodEndAt, |
| 273 | profile: snapshot.profile, |
| 274 | metrics: snapshot.metrics, |
| 275 | extra: snapshot.extra, |
| 276 | rawResponse: snapshot.rawResponse ?? rawResponse, |
| 277 | })), |
| 278 | ) |
| 279 | } |
| 280 | |
| 281 | private async saveWorkSnapshots( |
| 282 | userId: string, |
| 283 | platform: AccountType, |
| 284 | platformWorkId: string, |
| 285 | accountId: string | undefined, |
| 286 | snapshots: ChannelWorkDataSnapshotPayload[], |
| 287 | rawResponse?: unknown, |
| 288 | ) { |
| 289 | if (snapshots.length === 0) { |
| 290 | return [] |
| 291 | } |
| 292 | return this.workSnapshotRepository.createMany( |
| 293 | snapshots.map(snapshot => ({ |
| 294 | userId, |
| 295 | platform, |
| 296 | accountId, |
| 297 | platformWorkId: snapshot.platformWorkId ?? platformWorkId, |
| 298 | snapshotAt: snapshot.snapshotAt, |
| 299 | fetchedAt: snapshot.fetchedAt ?? new Date(), |
| 300 | periodStartAt: snapshot.periodStartAt, |
| 301 | periodEndAt: snapshot.periodEndAt, |
| 302 | work: snapshot.work, |
| 303 | metrics: snapshot.metrics, |
| 304 | extra: snapshot.extra, |
| 305 | rawResponse: snapshot.rawResponse ?? rawResponse, |
| 306 | })), |
| 307 | ) |
| 308 | } |
| 309 | |
| 310 | private async syncAccountMetrics(accountId: string, metrics: ChannelAccountMetricsSnapshot, fetchedAt: Date) { |
| 311 | const update: { |
| 312 | fansCount?: number |
| 313 | followingCount?: number |
| 314 | workCount?: number |
| 315 | readCount?: number |
| 316 | likeCount?: number |
| 317 | collectCount?: number |
| 318 | forwardCount?: number |
| 319 | commentCount?: number |
| 320 | lastStatsTime: Date |
| 321 | } = { |
| 322 | lastStatsTime: fetchedAt, |
| 323 | } |
| 324 | if (metrics.fansCount !== undefined) { |
| 325 | update.fansCount = metrics.fansCount |
| 326 | } |
| 327 | if (metrics.followingCount !== undefined) { |
| 328 | update.followingCount = metrics.followingCount |
| 329 | } |
| 330 | if (metrics.workCount !== undefined) { |
| 331 | update.workCount = metrics.workCount |
| 332 | } |
| 333 | if (metrics.readCount !== undefined) { |
| 334 | update.readCount = metrics.readCount |
| 335 | } |
| 336 | if (metrics.likeCount !== undefined) { |
| 337 | update.likeCount = metrics.likeCount |
| 338 | } |
| 339 | if (metrics.collectCount !== undefined) { |
| 340 | update.collectCount = metrics.collectCount |
| 341 | } |
| 342 | if (metrics.forwardCount !== undefined) { |
| 343 | update.forwardCount = metrics.forwardCount |
| 344 | } |
| 345 | if (metrics.commentCount !== undefined) { |
| 346 | update.commentCount = metrics.commentCount |
| 347 | } |
| 348 | await this.accountRepository.updateById(accountId, update) |
| 349 | } |
| 350 | } |
| 351 |