Spaces:
Runtime error
Runtime error
File size: 16,561 Bytes
ceb943f 304bdf5 ceb943f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 | 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;
}
}
);
|