| 1 | import type { AssetsConfig } from './assets.config' |
| 2 | import { Inject, Injectable, Logger } from '@nestjs/common' |
| 3 | import { AssetRepository, AssetStatus } from '@yikart/mongodb' |
| 4 | import { ASSETS_CONFIG } from './assets.config' |
| 5 | |
| 6 | export interface R2EventMessage { |
| 7 | account: string |
| 8 | bucket: string |
| 9 | object: { |
| 10 | key: string |
| 11 | size: number |
| 12 | eTag: string |
| 13 | } |
| 14 | action: 'PutObject' | 'CopyObject' | 'CompleteMultipartUpload' | 'DeleteObject' |
| 15 | eventTime: string |
| 16 | } |
| 17 | |
| 18 | export interface CloudflareQueueMessage { |
| 19 | body: R2EventMessage |
| 20 | id: string |
| 21 | timestamp_ms: number |
| 22 | attempts: number |
| 23 | lease_id: string |
| 24 | } |
| 25 | |
| 26 | @Injectable() |
| 27 | export class R2EventsService { |
| 28 | private readonly logger = new Logger(R2EventsService.name) |
| 29 | private readonly baseUrl: string |
| 30 | private readonly apiToken: string |
| 31 | |
| 32 | constructor( |
| 33 | @Inject(ASSETS_CONFIG) private readonly config: AssetsConfig, |
| 34 | private readonly assetRepository: AssetRepository, |
| 35 | ) { |
| 36 | const cf = config.provider === 's3' ? config.cloudflare : undefined |
| 37 | this.baseUrl = `https://api.cloudflare.com/client/v4/accounts/${cf?.accountId}/queues/${cf?.queueId}/messages` |
| 38 | this.apiToken = cf?.apiToken || '' |
| 39 | } |
| 40 | |
| 41 | async pullMessages(batchSize = 10): Promise<CloudflareQueueMessage[]> { |
| 42 | const response = await fetch(`${this.baseUrl}/pull`, { |
| 43 | method: 'POST', |
| 44 | headers: { |
| 45 | 'Authorization': `Bearer ${this.apiToken}`, |
| 46 | 'Content-Type': 'application/json', |
| 47 | }, |
| 48 | body: JSON.stringify({ batch_size: batchSize, visibility_timeout_ms: 30000 }), |
| 49 | }) |
| 50 | |
| 51 | const data = await response.json() as { |
| 52 | success: boolean |
| 53 | result: { messages: CloudflareQueueMessage[] } |
| 54 | } |
| 55 | |
| 56 | return data.result?.messages || [] |
| 57 | } |
| 58 | |
| 59 | async ackMessages(leaseIds: string[]): Promise<void> { |
| 60 | if (leaseIds.length === 0) |
| 61 | return |
| 62 | |
| 63 | await fetch(`${this.baseUrl}/ack`, { |
| 64 | method: 'POST', |
| 65 | headers: { |
| 66 | 'Authorization': `Bearer ${this.apiToken}`, |
| 67 | 'Content-Type': 'application/json', |
| 68 | }, |
| 69 | body: JSON.stringify({ acks: leaseIds.map(lease_id => ({ lease_id })) }), |
| 70 | }) |
| 71 | } |
| 72 | |
| 73 | async processMessages(): Promise<{ processed: number, failed: number }> { |
| 74 | const messages = await this.pullMessages(10) |
| 75 | if (messages.length === 0) |
| 76 | return { processed: 0, failed: 0 } |
| 77 | |
| 78 | const processedIds: string[] = [] |
| 79 | let failed = 0 |
| 80 | |
| 81 | for (const msg of messages) { |
| 82 | const event = msg.body |
| 83 | if (!['PutObject', 'CopyObject', 'CompleteMultipartUpload'].includes(event.action)) { |
| 84 | processedIds.push(msg.lease_id) |
| 85 | continue |
| 86 | } |
| 87 | |
| 88 | const asset = await this.assetRepository.getByPath(event.object.key) |
| 89 | if (!asset || asset.status !== AssetStatus.Pending) { |
| 90 | processedIds.push(msg.lease_id) |
| 91 | continue |
| 92 | } |
| 93 | |
| 94 | try { |
| 95 | await this.assetRepository.updateStatus(asset.id, AssetStatus.Confirmed, { |
| 96 | size: event.object.size, |
| 97 | }) |
| 98 | processedIds.push(msg.lease_id) |
| 99 | this.logger.log(`Asset confirmed via R2 event: ${event.object.key}`) |
| 100 | } |
| 101 | catch { |
| 102 | failed++ |
| 103 | } |
| 104 | } |
| 105 | |
| 106 | if (processedIds.length > 0) { |
| 107 | await this.ackMessages(processedIds) |
| 108 | } |
| 109 | |
| 110 | return { processed: processedIds.length, failed } |
| 111 | } |
| 112 | } |
| 113 |