import { inngest } from '../client'; import { extractFramesAtIntervals, VideoUnavailableError } from '@/features/discovery/services/frame-extraction.service'; import { detectObjectsInFrame, VisionRateLimitError } from '@/features/discovery/services/ai-vision.service'; import { db } from '@/lib/db'; import { detectedObjects, youtubeVideos } from '@/lib/db/schema'; import * as Sentry from '@sentry/nextjs'; import { NonRetriableError, RetryAfterError } from 'inngest'; import { eq } from 'drizzle-orm'; import sharp from 'sharp'; import { uploadThumbnail } from '@/features/discovery/services/storage.service'; export const detectObjects = inngest.createFunction( { id: 'detect-objects', retries: 3 }, { event: 'youtube/video.detect-objects' }, async ({ event, step }) => { const { videoId, videoUrl } = event.data; if (!videoId || !videoUrl) { throw new NonRetriableError('Missing videoId or videoUrl'); } try { // New: Mark as in_progress await step.run('update-scan-status-started', async () => { await db.update(youtubeVideos) .set({ scanStatus: 'in_progress' }) .where(eq(youtubeVideos.id, videoId)); }); // Step 1: Process Video (Extract & Detect) const detections = await step.run('analyze-video-content', async () => { try { // Calculate adaptive frame interval based on video duration // Short videos: more frequent sampling for better coverage // Long videos: less frequent to manage API costs const calculateFrameInterval = (durationSeconds: number | null | undefined): number => { const duration = durationSeconds || 300; // Default to 5 minutes if unknown if (duration <= 120) return 5; // 0-2 min: every 5 seconds if (duration <= 300) return 10; // 2-5 min: every 10 seconds if (duration <= 600) return 15; // 5-10 min: every 15 seconds return 30; // 10+ min: every 30 seconds }; // Get video metadata to determine duration const videoRecord = await db.query.youtubeVideos.findFirst({ where: eq(youtubeVideos.id, videoId) }); const durationSeconds = videoRecord?.duration ? parseInt(videoRecord.duration, 10) : 300; // Default to 5 minutes if unknown const frameInterval = calculateFrameInterval(durationSeconds); console.log(`Video duration: ${durationSeconds}s, using frame interval: ${frameInterval}s`); // Extract frames at adaptive intervals const frames = await extractFramesAtIntervals(videoUrl, frameInterval); const allResults: any[] = []; // Process frames in parallel batches of 3 for better performance const BATCH_SIZE = 3; const HEARTBEAT_INTERVAL = 5; // Update heartbeat every 5 frames for (let i = 0; i < frames.length; i += BATCH_SIZE) { const batch = frames.slice(i, i + BATCH_SIZE); // Process batch frames in parallel const batchResults = await Promise.allSettled( batch.map(async (frame) => { try { const results = await detectObjectsInFrame(frame.frameBuffer); // Process detections to add thumbnails const processedResults = await Promise.all( results.map(async (result) => { let thumbnailUrl = null; if (result.boundingBox) { try { // Get base image dimensions to convert normalized coordinates if needed const meta = await sharp(frame.frameBuffer).metadata(); const imgW = meta.width || 1280; const imgH = meta.height || 720; let { x, y, width, height } = result.boundingBox; // Detect coordinate format and normalize to pixel coordinates const maxCoord = Math.max(x, y, width, height); if (maxCoord <= 1) { // Normalized 0-1 coordinates x = x * imgW; y = y * imgH; width = width * imgW; height = height * imgH; } else if (maxCoord <= 1000 && maxCoord > imgW && maxCoord > imgH) { // Gemini 0-1000 scale x = (x / 1000) * imgW; y = (y / 1000) * imgH; width = (width / 1000) * imgW; height = (height / 1000) * imgH; } // else: already in pixel coordinates (HuggingFace format) // Add 20% padding around the object for better context const paddingX = width * 0.2; const paddingY = height * 0.2; const paddedX = x - paddingX; const paddedY = y - paddingY; const paddedWidth = width + (2 * paddingX); const paddedHeight = height + (2 * paddingY); // Clamp to image bounds const left = Math.max(0, Math.floor(paddedX)); const top = Math.max(0, Math.floor(paddedY)); const right = Math.min(imgW, Math.ceil(paddedX + paddedWidth)); const bottom = Math.min(imgH, Math.ceil(paddedY + paddedHeight)); const finalWidth = right - left; const finalHeight = bottom - top; // Validate dimensions (minimum 20x20 pixels) if (finalWidth >= 20 && finalHeight >= 20) { console.log(`Cropping ${result.name}: [${left},${top}] ${finalWidth}x${finalHeight} from ${imgW}x${imgH}`); const thumbnailBuffer = await sharp(frame.frameBuffer) .extract({ left, top, width: finalWidth, height: finalHeight }) .resize(300, 300, { fit: 'contain', background: { r: 255, g: 255, b: 255, alpha: 1 } }) .toFormat('jpeg', { quality: 85 }) .toBuffer(); const fileName = `${videoId}/${frame.timestamp}_${result.name.replace(/[^a-z0-9]/gi, '_').toLowerCase()}_${Date.now()}.jpg`; thumbnailUrl = await uploadThumbnail(thumbnailBuffer, fileName); console.log(`✓ Thumbnail saved: ${fileName}`); } else { throw new Error(`Region too small (${finalWidth}x${finalHeight})`); } } catch (cropError) { console.warn(`⚠ Thumbnail generation issue for ${result.name} (falling back to full frame):`, cropError); try { const fallbackBuffer = await sharp(frame.frameBuffer) .resize(300, 300, { fit: 'contain', background: { r: 255, g: 255, b: 255, alpha: 1 } }) .toFormat('jpeg', { quality: 85 }) .toBuffer(); const fallbackName = `${videoId}/${frame.timestamp}_${result.name.replace(/[^a-z0-9]/gi, '_').toLowerCase()}_full_${Date.now()}.jpg`; thumbnailUrl = await uploadThumbnail(fallbackBuffer, fallbackName); console.log(`✓ Fallback thumbnail saved: ${fallbackName}`); } catch (fallbackError) { console.error(`✗ Fallback thumbnail generation failed for ${result.name}:`, fallbackError); } } } return { ...result, timestamp: frame.timestamp, thumbnailUrl }; }) ); return processedResults; } catch (err) { // Handle rate limiting if (err instanceof VisionRateLimitError) { throw new RetryAfterError( `Vision API rate limited`, `${err.retryAfter || 60}s` ); } // Log but continue processing other frames Sentry.captureException(err, { tags: { source: 'vision-detection', videoId, frameTimestamp: frame.timestamp, }, }); return []; // Return empty array for failed frames } }) ); // Collect successful results from the batch for (const result of batchResults) { if (result.status === 'fulfilled') { allResults.push(...result.value); } } // Periodic heartbeat update (every HEARTBEAT_INTERVAL frames, not per frame) if (i % HEARTBEAT_INTERVAL === 0) { await db.update(youtubeVideos) .set({ updatedAt: new Date() }) .where(eq(youtubeVideos.id, videoId)); } } return allResults; } catch (err) { // Propagate rate limit errors if (err instanceof RetryAfterError) throw err; throw err; } }); // Step 2: Save Results (filter out Person category - only save products) const savedObjectIds = await step.run('save-detections', async () => { // Filter out Person detections - only interested in products const productDetections = detections.filter((d: any) => d.category !== 'Person'); if (productDetections.length === 0) return []; const records = productDetections.map((d: any) => ({ videoId, objectName: d.name, category: d.category, confidenceScore: d.confidence, frameTimestamp: d.timestamp, detectionMetadata: d.boundingBox ? { boundingBox: d.boundingBox } : null, thumbnailUrl: d.thumbnailUrl, status: (d.confidence < 0.7 ? 'flagged' : 'pending_review') as 'flagged' | 'pending_review' })); const results = await db.insert(detectedObjects).values(records).returning({ id: detectedObjects.id }); return results.map(r => r.id); }); // Step 3: Trigger Marketplace Match (single batch event for all objects) if (savedObjectIds.length > 0) { await step.run('trigger-marketplace-match', async () => { await inngest.send({ name: 'discovery/objects.match-marketplace', data: { detectedObjectIds: savedObjectIds, videoId } }); }); } await step.run('update-scan-status-finished', async () => { await db.update(youtubeVideos) .set({ scanStatus: 'awaiting_approval' }) .where(eq(youtubeVideos.id, videoId)); }); return { success: true, count: detections.length }; } catch (error) { // New: Mark as failed await step.run('update-scan-status-failed', async () => { await db.update(youtubeVideos) .set({ scanStatus: 'failed' }) .where(eq(youtubeVideos.id, videoId)); }); if (error instanceof VideoUnavailableError) { console.warn(`Skipping video ${videoId}: ${error.message}`); await db.update(youtubeVideos) .set({ availabilityStatus: 'private' }) .where(eq(youtubeVideos.id, videoId)); throw new NonRetriableError(error.message); } Sentry.captureException(error, { tags: { source: 'inngest', function: 'detect-objects', videoId } }); throw error; } } );