| 1 | import type { Readable } from 'node:stream' |
| 2 | import { Inject, Injectable, Logger } from '@nestjs/common' |
| 3 | import { AppException, ResponseCode, UserType } from '@yikart/common' |
| 4 | import { Asset, AssetRepository, AssetStatus, AssetType } from '@yikart/mongodb' |
| 5 | import * as mime from 'mime-types' |
| 6 | import { ASSETS_CONFIG, AssetsConfig } from './assets.config' |
| 7 | import { |
| 8 | ConfirmAssetDto, |
| 9 | ListAssetsDto, |
| 10 | UploadAssetDto, |
| 11 | UploadFromBufferDto, |
| 12 | UploadFromUrlDto, |
| 13 | } from './dto' |
| 14 | import { StorageProvider } from './storage-provider' |
| 15 | import { generateAssetPath, generateAssetPathFromFilename, PathGeneratorOptions } from './utils/path-generator' |
| 16 | import { VideoMetadataService } from './video-metadata.service' |
| 17 | |
| 18 | export interface UploadResult { |
| 19 | asset: Asset |
| 20 | url: string |
| 21 | uploadUrl?: string |
| 22 | } |
| 23 | |
| 24 | @Injectable() |
| 25 | export class AssetsService { |
| 26 | private readonly logger = new Logger(AssetsService.name) |
| 27 | |
| 28 | constructor( |
| 29 | private readonly storage: StorageProvider, |
| 30 | private readonly assetRepository: AssetRepository, |
| 31 | private readonly videoMetadataService: VideoMetadataService, |
| 32 | @Inject(ASSETS_CONFIG) protected readonly options: AssetsConfig, |
| 33 | ) {} |
| 34 | |
| 35 | async createUploadSign( |
| 36 | userId: string, |
| 37 | dto: UploadAssetDto, |
| 38 | userType: UserType = UserType.User, |
| 39 | ): Promise<Required<UploadResult>> { |
| 40 | if (this.options.maxSize != null && dto.size && dto.size >= this.options.maxSize) { |
| 41 | throw new AppException(ResponseCode.AssetTooLarge) |
| 42 | } |
| 43 | const pathOptions: PathGeneratorOptions = { |
| 44 | userId, |
| 45 | userType, |
| 46 | type: dto.type, |
| 47 | mimeType: dto.mimeType, |
| 48 | filename: dto.filename, |
| 49 | } |
| 50 | |
| 51 | const path = generateAssetPath(pathOptions) |
| 52 | |
| 53 | const asset = await this.assetRepository.create({ |
| 54 | userId, |
| 55 | userType, |
| 56 | path, |
| 57 | type: dto.type, |
| 58 | status: AssetStatus.Pending, |
| 59 | size: dto.size, |
| 60 | mimeType: dto.mimeType, |
| 61 | filename: dto.filename, |
| 62 | metadata: dto.metadata, |
| 63 | expiresAt: dto.expiresInSeconds |
| 64 | ? new Date(Date.now() + dto.expiresInSeconds * 1000) |
| 65 | : undefined, |
| 66 | }) |
| 67 | |
| 68 | const signUrl = await this.storage.getUploadSignUrl(path, dto.mimeType, dto.size, { assetId: asset.id }) |
| 69 | |
| 70 | return { |
| 71 | asset, |
| 72 | url: this.buildUrl(path), |
| 73 | uploadUrl: signUrl, |
| 74 | } |
| 75 | } |
| 76 | |
| 77 | async uploadFromUrl( |
| 78 | userId: string, |
| 79 | dto: UploadFromUrlDto, |
| 80 | subPath?: string, |
| 81 | userType: UserType = UserType.User, |
| 82 | ): Promise<UploadResult> { |
| 83 | const filename = dto.filename || new URL(dto.url).pathname.split('/').pop() || undefined |
| 84 | |
| 85 | const pathOptions: PathGeneratorOptions = { |
| 86 | userId, |
| 87 | userType, |
| 88 | type: dto.type, |
| 89 | mimeType: 'application/octet-stream', |
| 90 | filename, |
| 91 | subPath, |
| 92 | } |
| 93 | |
| 94 | const tempPath = generateAssetPathFromFilename(pathOptions) |
| 95 | |
| 96 | const result = await this.storage.putObjectFromUrl(dto.url, tempPath) |
| 97 | |
| 98 | const headResult = await this.storage.headObject(result.path) |
| 99 | |
| 100 | const asset = await this.assetRepository.create({ |
| 101 | userId, |
| 102 | userType, |
| 103 | path: result.path, |
| 104 | type: dto.type, |
| 105 | status: AssetStatus.Confirmed, |
| 106 | size: headResult.contentLength || 0, |
| 107 | mimeType: headResult.contentType || 'application/octet-stream', |
| 108 | filename: dto.filename, |
| 109 | metadata: dto.metadata, |
| 110 | }) |
| 111 | |
| 112 | return { |
| 113 | asset, |
| 114 | url: this.buildUrl(result.path), |
| 115 | } |
| 116 | } |
| 117 | |
| 118 | async uploadFromBuffer( |
| 119 | userId: string, |
| 120 | buffer: Buffer, |
| 121 | dto: UploadFromBufferDto, |
| 122 | subPath?: string, |
| 123 | userType: UserType = UserType.User, |
| 124 | ): Promise<UploadResult> { |
| 125 | const pathOptions: PathGeneratorOptions = { |
| 126 | userId, |
| 127 | userType, |
| 128 | type: dto.type, |
| 129 | mimeType: dto.mimeType, |
| 130 | filename: dto.filename, |
| 131 | subPath, |
| 132 | } |
| 133 | |
| 134 | const path = generateAssetPath(pathOptions) |
| 135 | |
| 136 | await this.storage.putObject(path, buffer, dto.mimeType) |
| 137 | |
| 138 | const asset = await this.assetRepository.create({ |
| 139 | userId, |
| 140 | userType, |
| 141 | path, |
| 142 | type: dto.type, |
| 143 | status: AssetStatus.Confirmed, |
| 144 | size: buffer.length, |
| 145 | mimeType: dto.mimeType, |
| 146 | filename: dto.filename, |
| 147 | metadata: dto.metadata, |
| 148 | }) |
| 149 | |
| 150 | return { |
| 151 | asset, |
| 152 | url: this.buildUrl(path), |
| 153 | } |
| 154 | } |
| 155 | |
| 156 | async uploadFromStream( |
| 157 | userId: string, |
| 158 | stream: Buffer | Readable, |
| 159 | dto: UploadFromBufferDto & { size: number }, |
| 160 | subPath?: string, |
| 161 | userType: UserType = UserType.User, |
| 162 | ): Promise<UploadResult> { |
| 163 | const pathOptions: PathGeneratorOptions = { |
| 164 | userId, |
| 165 | userType, |
| 166 | type: dto.type, |
| 167 | mimeType: dto.mimeType, |
| 168 | filename: dto.filename, |
| 169 | subPath, |
| 170 | } |
| 171 | |
| 172 | const path = generateAssetPath(pathOptions) |
| 173 | |
| 174 | await this.storage.putObject(path, stream, dto.mimeType) |
| 175 | |
| 176 | const asset = await this.assetRepository.create({ |
| 177 | userId, |
| 178 | userType, |
| 179 | path, |
| 180 | type: dto.type, |
| 181 | status: AssetStatus.Confirmed, |
| 182 | size: dto.size, |
| 183 | mimeType: dto.mimeType, |
| 184 | filename: dto.filename, |
| 185 | metadata: dto.metadata, |
| 186 | }) |
| 187 | |
| 188 | return { |
| 189 | asset, |
| 190 | url: this.buildUrl(path), |
| 191 | } |
| 192 | } |
| 193 | |
| 194 | async confirmUpload(dto: ConfirmAssetDto): Promise<Asset> { |
| 195 | if (this.options.maxSize != null && dto.size && dto.size >= this.options.maxSize) { |
| 196 | throw new AppException(ResponseCode.AssetTooLarge) |
| 197 | } |
| 198 | const asset = await this.assetRepository.getById(dto.assetId) |
| 199 | |
| 200 | if (!asset) { |
| 201 | throw new AppException(ResponseCode.AssetNotFound) |
| 202 | } |
| 203 | |
| 204 | if (asset.status !== AssetStatus.Pending) { |
| 205 | return asset |
| 206 | } |
| 207 | |
| 208 | const updateData: Partial<Asset> = { |
| 209 | status: AssetStatus.Confirmed, |
| 210 | } |
| 211 | |
| 212 | if (dto.size) { |
| 213 | updateData.size = dto.size |
| 214 | } |
| 215 | |
| 216 | if (dto.metadata) { |
| 217 | updateData.metadata = { ...asset.metadata, ...dto.metadata } |
| 218 | } |
| 219 | |
| 220 | const updated = await this.assetRepository.updateById(dto.assetId, updateData) |
| 221 | return updated! |
| 222 | } |
| 223 | |
| 224 | async getById(assetId: string): Promise<Asset | null> { |
| 225 | return await this.assetRepository.getById(assetId) |
| 226 | } |
| 227 | |
| 228 | async getByPath(path: string): Promise<Asset | null> { |
| 229 | return await this.assetRepository.getByPath(path) |
| 230 | } |
| 231 | |
| 232 | parsePathFromUrl(url: string): string { |
| 233 | return this.storage.parsePathFromUrl(url) |
| 234 | } |
| 235 | |
| 236 | async toPresignedUrl(urlOrPath: string, expiresIn = 3600): Promise<string> { |
| 237 | return this.storage.toPresignedUrl(urlOrPath, expiresIn) |
| 238 | } |
| 239 | |
| 240 | async getOrCreateAssetByPath(path: string, userId: string, userType: UserType = UserType.User): Promise<Asset> { |
| 241 | const existingAsset = await this.assetRepository.getByPath(path) |
| 242 | if (existingAsset) { |
| 243 | return existingAsset |
| 244 | } |
| 245 | |
| 246 | let headResult |
| 247 | try { |
| 248 | headResult = await this.storage.headObject(path) |
| 249 | } |
| 250 | catch { |
| 251 | throw new AppException(ResponseCode.AssetNotFound, 'File not found in storage') |
| 252 | } |
| 253 | |
| 254 | if (!headResult) { |
| 255 | throw new AppException(ResponseCode.AssetNotFound, 'File not found in storage') |
| 256 | } |
| 257 | |
| 258 | const asset = await this.assetRepository.create({ |
| 259 | userId, |
| 260 | userType, |
| 261 | path, |
| 262 | type: AssetType.Temp, |
| 263 | status: AssetStatus.Confirmed, |
| 264 | size: headResult.contentLength || 0, |
| 265 | mimeType: headResult.contentType || 'application/octet-stream', |
| 266 | }) |
| 267 | |
| 268 | return asset |
| 269 | } |
| 270 | |
| 271 | async listWithPagination(userId: string, dto: ListAssetsDto, userType: UserType = UserType.User) { |
| 272 | return await this.assetRepository.listWithPagination({ |
| 273 | userId, |
| 274 | userType, |
| 275 | page: dto.page, |
| 276 | pageSize: dto.pageSize, |
| 277 | type: dto.type, |
| 278 | }) |
| 279 | } |
| 280 | |
| 281 | async softDelete(userId: string, assetId: string, userType: UserType = UserType.User): Promise<void> { |
| 282 | const asset = await this.assetRepository.getByIdAndUserId(assetId, userId, userType) |
| 283 | |
| 284 | if (!asset) { |
| 285 | throw new AppException(ResponseCode.AssetNotFound) |
| 286 | } |
| 287 | |
| 288 | await this.assetRepository.softDelete(assetId) |
| 289 | } |
| 290 | |
| 291 | buildUrl(path: string): string { |
| 292 | return this.storage.buildUrl(path) |
| 293 | } |
| 294 | |
| 295 | async processPendingAssets(olderThanSeconds: number, limit = 100): Promise<{ confirmed: number, failed: number }> { |
| 296 | const pendingAssets = await this.assetRepository.findPendingAssets(olderThanSeconds, limit) |
| 297 | let confirmed = 0 |
| 298 | let failed = 0 |
| 299 | |
| 300 | for (const asset of pendingAssets) { |
| 301 | try { |
| 302 | const exists = await this.storage.headObject(asset.path) |
| 303 | if (exists) { |
| 304 | await this.assetRepository.updateStatus(asset.id, AssetStatus.Confirmed, { |
| 305 | size: exists.contentLength, |
| 306 | }) |
| 307 | confirmed++ |
| 308 | } |
| 309 | else { |
| 310 | await this.assetRepository.updateStatus(asset.id, AssetStatus.Failed) |
| 311 | failed++ |
| 312 | } |
| 313 | } |
| 314 | catch { |
| 315 | await this.assetRepository.updateStatus(asset.id, AssetStatus.Failed) |
| 316 | failed++ |
| 317 | } |
| 318 | } |
| 319 | |
| 320 | return { confirmed, failed } |
| 321 | } |
| 322 | |
| 323 | async processQuickPollPendingAssets(limit = 50): Promise<{ confirmed: number }> { |
| 324 | const pendingAssets = await this.assetRepository.listByStatusAndAge(AssetStatus.Pending, 5, limit) |
| 325 | let confirmed = 0 |
| 326 | |
| 327 | for (const asset of pendingAssets) { |
| 328 | try { |
| 329 | const exists = await this.storage.headObject(asset.path) |
| 330 | if (exists) { |
| 331 | await this.assetRepository.updateStatus(asset.id, AssetStatus.Confirmed, { |
| 332 | size: exists.contentLength, |
| 333 | }) |
| 334 | confirmed++ |
| 335 | } |
| 336 | } |
| 337 | catch { |
| 338 | // headObject 失败，文件可能还未上传完成，不做处理 |
| 339 | } |
| 340 | } |
| 341 | |
| 342 | return { confirmed } |
| 343 | } |
| 344 | |
| 345 | async cleanupExpiredPendingAssets(olderThanSeconds: number): Promise<{ affectedCount: number }> { |
| 346 | return await this.assetRepository.markExpiredPendingAsFailed(olderThanSeconds) |
| 347 | } |
| 348 | |
| 349 | async confirmUploadByUser(userId: string, assetId: string, userType: UserType = UserType.User): Promise<Asset> { |
| 350 | const asset = await this.assetRepository.getByIdAndUserId(assetId, userId, userType) |
| 351 | |
| 352 | if (!asset) { |
| 353 | throw new AppException(ResponseCode.AssetNotFound) |
| 354 | } |
| 355 | |
| 356 | if (asset.status !== AssetStatus.Pending) { |
| 357 | return asset |
| 358 | } |
| 359 | |
| 360 | const headResult = await this.storage.headObject(asset.path) |
| 361 | if (!headResult) { |
| 362 | throw new AppException(ResponseCode.AssetUploadFailed) |
| 363 | } |
| 364 | |
| 365 | if (this.options.maxSize != null && headResult.contentLength && headResult.contentLength >= this.options.maxSize) { |
| 366 | throw new AppException(ResponseCode.AssetTooLarge) |
| 367 | } |
| 368 | |
| 369 | const expectedMimeType = asset.filename |
| 370 | ? (mime.lookup(asset.filename) || asset.mimeType) |
| 371 | : asset.mimeType |
| 372 | const currentContentType = headResult.contentType |
| 373 | |
| 374 | if (expectedMimeType && currentContentType !== expectedMimeType) { |
| 375 | const contentDisposition = expectedMimeType.startsWith('video/') ? 'inline' : undefined |
| 376 | await this.storage.copyObject(asset.path, { |
| 377 | contentType: expectedMimeType, |
| 378 | contentDisposition, |
| 379 | metadata: { |
| 380 | assetId: asset.id, |
| 381 | userId: asset.userId, |
| 382 | }, |
| 383 | }) |
| 384 | } |
| 385 | |
| 386 | const finalMimeType = expectedMimeType || asset.mimeType |
| 387 | |
| 388 | const updateData: Partial<Asset> = { |
| 389 | status: AssetStatus.Confirmed, |
| 390 | size: headResult.contentLength || asset.size, |
| 391 | mimeType: finalMimeType, |
| 392 | } |
| 393 | |
| 394 | if (finalMimeType?.startsWith('video/')) { |
| 395 | try { |
| 396 | const videoUrl = this.storage.buildUrl(asset.path) |
| 397 | const videoMetadata = await this.videoMetadataService.probeVideoMetadata(videoUrl) |
| 398 | updateData.metadata = { |
| 399 | ...asset.metadata, |
| 400 | width: videoMetadata.width, |
| 401 | height: videoMetadata.height, |
| 402 | duration: videoMetadata.duration, |
| 403 | bitrate: videoMetadata.bitrate, |
| 404 | frameRate: videoMetadata.frameRate, |
| 405 | } |
| 406 | this.logger.debug({ assetId, videoMetadata }, 'Video metadata extracted successfully') |
| 407 | } |
| 408 | catch (error) { |
| 409 | this.logger.warn({ assetId, error }, 'Failed to extract video metadata, skipping') |
| 410 | } |
| 411 | } |
| 412 | |
| 413 | const updated = await this.assetRepository.updateById(assetId, updateData) |
| 414 | |
| 415 | return updated! |
| 416 | } |
| 417 | } |
| 418 |