diff --git a/backend/alembic/versions/a1b2c3d4e5f6_add_media_analysis_table.py b/backend/alembic/versions/a1b2c3d4e5f6_add_media_analysis_table.py new file mode 100644 index 00000000..7bd65738 --- /dev/null +++ b/backend/alembic/versions/a1b2c3d4e5f6_add_media_analysis_table.py @@ -0,0 +1,42 @@ +"""Add media_analysis table with pgvector embedding + +Revision ID: a1b2c3d4e5f6 +Revises: e8a1b2c3d4f5 +Create Date: 2026-07-28 00:00:00.000000 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +revision: str = 'a1b2c3d4e5f6' +down_revision: Union[str, Sequence[str], None] = 'e8a1b2c3d4f5' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.execute("CREATE EXTENSION IF NOT EXISTS vector") + + op.create_table( + 'media_analysis', + sa.Column('id', sa.Integer(), primary_key=True, autoincrement=True), + sa.Column('video_id', sa.Integer(), sa.ForeignKey('videos.id', ondelete='CASCADE'), nullable=False, unique=True), + sa.Column('status', sa.Text(), nullable=False, server_default='pending'), + sa.Column('frame_sample_paths', postgresql.JSONB(), nullable=True), + sa.Column('scene_cuts', postgresql.JSONB(), nullable=True), + sa.Column('beat_timestamps', postgresql.JSONB(), nullable=True), + sa.Column('caption_embedding', postgresql.dialects.postgresql.VECTOR(384), nullable=True), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.func.now(), onupdate=sa.func.now()), + ) + op.create_index('ix_media_analysis_video_id', 'media_analysis', ['video_id']) + op.create_index('ix_media_analysis_status', 'media_analysis', ['status']) + + +def downgrade() -> None: + op.drop_index('ix_media_analysis_status', table_name='media_analysis') + op.drop_index('ix_media_analysis_video_id', table_name='media_analysis') + op.drop_table('media_analysis') diff --git a/backend/alembic/versions/e8a1b2c3d4f5_add_video_events_table.py b/backend/alembic/versions/e8a1b2c3d4f5_add_video_events_table.py new file mode 100644 index 00000000..15aa3474 --- /dev/null +++ b/backend/alembic/versions/e8a1b2c3d4f5_add_video_events_table.py @@ -0,0 +1,41 @@ +"""Add video_events table for engagement tracking + +Revision ID: e8a1b2c3d4f5 +Revises: 7acd76e1bfde +Create Date: 2026-07-27 00:00:00.000000 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = 'e8a1b2c3d4f5' +down_revision: Union[str, Sequence[str], None] = '7acd76e1bfde' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + 'video_events', + sa.Column('id', sa.BigInteger(), primary_key=True, autoincrement=True), + sa.Column('user_id', sa.Integer(), sa.ForeignKey('users.id', ondelete='CASCADE'), nullable=False), + sa.Column('video_id', sa.Integer(), sa.ForeignKey('videos.id', ondelete='CASCADE'), nullable=False), + sa.Column('event_type', sa.Text(), nullable=False), + sa.Column('watch_seconds', sa.Integer(), nullable=True), + sa.Column('session_id', sa.Text(), nullable=True), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now()), + ) + op.create_index('ix_video_events_user_created', 'video_events', ['user_id', 'created_at']) + op.create_index('ix_video_events_video_event', 'video_events', ['video_id', 'event_type']) + op.create_index('ix_video_events_video_created', 'video_events', ['video_id', 'created_at']) + + +def downgrade() -> None: + op.drop_index('ix_video_events_video_created', table_name='video_events') + op.drop_index('ix_video_events_video_event', table_name='video_events') + op.drop_index('ix_video_events_user_created', table_name='video_events') + op.drop_table('video_events') diff --git a/backend/app/api/v1/api.py b/backend/app/api/v1/api.py index b5902cad..a8937bcf 100644 --- a/backend/app/api/v1/api.py +++ b/backend/app/api/v1/api.py @@ -1,5 +1,5 @@ from fastapi import APIRouter -from app.api.v1.endpoints import auth, videos, admin, users, posts, achievements, notifications, ads, chat, challenges, monetization, video_views, seo, recommendations, partners, categories, library, history, watch_later, liked, following, metrics, email, reports, comments +from app.api.v1.endpoints import auth, videos, admin, users, posts, achievements, notifications, ads, chat, challenges, monetization, video_views, seo, recommendations, partners, categories, library, history, watch_later, liked, following, metrics, email, reports, comments, events api_router = APIRouter() @@ -28,6 +28,7 @@ api_router.include_router(email.router, prefix="/email", tags=["email"]) api_router.include_router(reports.router, prefix="/reports", tags=["reports"]) api_router.include_router(comments.router, prefix="/comments", tags=["comments"]) +api_router.include_router(events.router, prefix="/events", tags=["events"]) diff --git a/backend/app/api/v1/endpoints/events.py b/backend/app/api/v1/endpoints/events.py new file mode 100644 index 00000000..3ed261d3 --- /dev/null +++ b/backend/app/api/v1/endpoints/events.py @@ -0,0 +1,39 @@ +import json +import logging +from typing import List, Union + +from fastapi import APIRouter, Response +from app.schemas.video_events import VideoEventIn, VideoEventBatch +from app.core.redis import redis_client + +logger = logging.getLogger(__name__) + +router = APIRouter() + +PENDING_KEY = "video_events:pending" + + +@router.post("/video", status_code=202) +def ingest_video_events(payload: Union[VideoEventBatch, List[VideoEventIn]]): + """Accept one or a batch of video engagement events and push to Redis for async processing.""" + events = payload.events if isinstance(payload, VideoEventBatch) else payload + + if not events: + return Response(status_code=202) + + pipe = redis_client.pipeline() + for ev in events: + pipe.rpush( + PENDING_KEY, + json.dumps({ + "user_id": ev.user_id, + "video_id": ev.video_id, + "event_type": ev.event_type.value, + "watch_seconds": ev.watch_seconds, + "session_id": ev.session_id, + }), + ) + pipe.execute() + + logger.debug("Enqueued %d video events to Redis", len(events)) + return Response(status_code=202) diff --git a/backend/app/api/v1/endpoints/videos.py b/backend/app/api/v1/endpoints/videos.py index 1a7d1455..3819c880 100644 --- a/backend/app/api/v1/endpoints/videos.py +++ b/backend/app/api/v1/endpoints/videos.py @@ -1510,3 +1510,20 @@ def delete_comment( if result is False: raise HTTPException(status_code=403, detail="Not authorized to delete this comment") return {"status": "success", "message": "Comment deleted successfully"} + + +# ── Media Analysis ──────────────────────────────────────────────────────────── + +@router.get("/{video_id}/analysis") +def get_media_analysis(video_id: str, db: Session = Depends(get_db)): + """Return the media_analysis row for a video (frames, scene cuts, beats, embedding).""" + from sqlalchemy import text + row = db.execute( + text("SELECT * FROM media_analysis WHERE video_id = :vid"), + {"vid": video_id}, + ).mappings().first() + + if not row: + raise HTTPException(status_code=404, detail="No analysis found for this video") + + return dict(row) diff --git a/backend/app/schemas/video_events.py b/backend/app/schemas/video_events.py new file mode 100644 index 00000000..0e50f82d --- /dev/null +++ b/backend/app/schemas/video_events.py @@ -0,0 +1,36 @@ +from pydantic import BaseModel, field_validator +from typing import Optional, List, Union +import enum + + +class VideoEventType(str, enum.Enum): + VIEW = "view" + WATCH_25 = "watch_25" + WATCH_50 = "watch_50" + WATCH_75 = "watch_75" + COMPLETE = "complete" + LIKE = "like" + SHARE = "share" + SKIP = "skip" + + +class VideoEventIn(BaseModel): + user_id: int + video_id: int + event_type: VideoEventType + watch_seconds: Optional[int] = None + session_id: Optional[str] = None + + @field_validator("event_type", mode="before") + @classmethod + def coerce_event_type(cls, v): + if isinstance(v, VideoEventType): + return v + try: + return VideoEventType(v) + except ValueError: + raise ValueError(f"Invalid event_type: {v}. Must be one of: {[e.value for e in VideoEventType]}") + + +class VideoEventBatch(BaseModel): + events: List[VideoEventIn] diff --git a/backend/app/services/media_analysis.py b/backend/app/services/media_analysis.py new file mode 100644 index 00000000..e4789c11 --- /dev/null +++ b/backend/app/services/media_analysis.py @@ -0,0 +1,133 @@ +""" +Media analysis pipeline — extracts frames, scene cuts, beat timestamps, +and caption embeddings from a transcoded video. + +Called by the Celery task after transcoding completes. +""" +import os +import tempfile +import shutil +import logging +import subprocess +import json +from typing import List, Optional + +import numpy as np + +logger = logging.getLogger(__name__) + + +def extract_frames(video_path: str, video_id: str) -> List[str]: + """Sample 1 frame per second from the video, upload each to S3, return S3 keys.""" + from app.core.storage import storage + + frames_dir = tempfile.mkdtemp(prefix="frames_") + try: + subprocess.run( + [ + "ffmpeg", "-i", video_path, + "-vf", "fps=1", + "-q:v", "3", + os.path.join(frames_dir, "frame_%04d.jpg"), + ], + check=True, + capture_output=True, + timeout=600, + ) + + s3_keys = [] + for fname in sorted(os.listdir(frames_dir)): + local_path = os.path.join(frames_dir, fname) + s3_key = f"analysis/{video_id}/frames/{fname}" + storage.upload_file(local_path, s3_key) + s3_keys.append(s3_key) + + logger.info("Uploaded %d frames for video %s", len(s3_keys), video_id) + return s3_keys + finally: + shutil.rmtree(frames_dir, ignore_errors=True) + + +def detect_scene_cuts(video_path: str) -> List[float]: + """Detect scene cuts using pyscenedetect ContentDetector. Returns timestamps in seconds.""" + from scenedetect import open_video, SceneManager + from scenedetect.detectors import ContentDetector + + video = open_video(video_path) + scene_manager = SceneManager() + scene_manager.add_detector(ContentDetector(threshold=27.0)) + scene_manager.detect_scenes(video, show_progress=False) + + scene_list = scene_manager.get_scene_list() + # Return the start time of each scene as seconds + timestamps = [round(scene[0].get_seconds(), 3) for scene in scene_list] + logger.info("Detected %d scene cuts", len(timestamps)) + return timestamps + + +def detect_beats(video_path: str) -> List[float]: + """Extract audio and detect beat timestamps using librosa. Returns timestamps in seconds.""" + import librosa + + # Extract audio to a temp WAV file + audio_tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False) + audio_tmp.close() + try: + subprocess.run( + [ + "ffmpeg", "-i", video_path, + "-vn", "-acodec", "pcm_s16le", "-ar", "22050", "-ac", "1", + "-y", audio_tmp.name, + ], + check=True, + capture_output=True, + timeout=300, + ) + + y, sr = librosa.load(audio_tmp.name, sr=22050) + tempo, beat_frames = librosa.beat.beat_track(y=y, sr=sr) + beat_times = librosa.frames_to_time(beat_frames, sr=sr).tolist() + beat_times = [round(t, 3) for t in beat_times] + logger.info("Detected %d beat timestamps", len(beat_times)) + return beat_times + finally: + os.unlink(audio_tmp.name) + + +def generate_caption_embedding(title: str, tags: Optional[str]) -> Optional[List[float]]: + """Generate a 384-dim embedding from title+tags using all-MiniLM-L6-v2.""" + from sentence_transformers import SentenceTransformer + + text = title or "" + if tags: + tag_str = " ".join(t.strip() for t in tags.split(",") if t.strip()) + text = f"{text} {tag_str}" + + text = text.strip() + if not text: + return None + + model = SentenceTransformer("all-MiniLM-L6-v2") + embedding = model.encode(text, normalize_embeddings=True) + return embedding.tolist() + + +def download_video_from_s3(video_s3_key: str) -> str: + """Download a video from S3 to a temp path. Returns the local file path.""" + from app.core.storage import storage + from app.core import config + + suffix = os.path.splitext(video_s3_key)[-1] or ".mp4" + tmp = tempfile.NamedTemporaryFile(suffix=suffix, delete=False) + tmp.close() + + try: + storage.s3_client.download_file( + config.AWS_STORAGE_BUCKET_NAME, + video_s3_key, + tmp.name, + ) + return tmp.name + except Exception: + os.unlink(tmp.name) + raise diff --git a/backend/app/tasks/event_tasks.py b/backend/app/tasks/event_tasks.py new file mode 100644 index 00000000..e205ba54 --- /dev/null +++ b/backend/app/tasks/event_tasks.py @@ -0,0 +1,55 @@ +import json +import logging +from datetime import datetime, timezone + +from celery import shared_task +from sqlalchemy import text + +# Bind @shared_task to the Redis-backed worker app +import app.worker # noqa: F401 +from app.db.session import SessionLocal +from app.core.redis import redis_client + +logger = logging.getLogger(__name__) + +PENDING_KEY = "video_events:pending" +FAILED_KEY = "video_events:failed" +BATCH_SIZE = 500 + + +@shared_task(name="tasks.events.flush_video_events") +def flush_video_events(): + """LPOP up to 500 pending events from Redis and bulk-insert into video_events.""" + events = [] + for _ in range(BATCH_SIZE): + raw = redis_client.lpop(PENDING_KEY) + if raw is None: + break + try: + events.append(json.loads(raw)) + except (json.JSONDecodeError, TypeError): + logger.warning("Skipping malformed event: %s", raw) + + if not events: + return + + db = SessionLocal() + try: + db.execute( + text( + "INSERT INTO video_events (user_id, video_id, event_type, watch_seconds, session_id, created_at) " + "VALUES (:user_id, :video_id, :event_type, :watch_seconds, :session_id, now())" + ), + events, + ) + db.commit() + logger.info("Flushed %d video events to database", len(events)) + except Exception as e: + db.rollback() + logger.error("Failed to flush %d video events: %s", len(events), e) + pipe = redis_client.pipeline() + for ev in events: + pipe.rpush(FAILED_KEY, json.dumps(ev)) + pipe.execute() + finally: + db.close() diff --git a/backend/app/tasks/media_tasks.py b/backend/app/tasks/media_tasks.py new file mode 100644 index 00000000..8a4c588f --- /dev/null +++ b/backend/app/tasks/media_tasks.py @@ -0,0 +1,141 @@ +import logging +import os +import tempfile + +from celery import shared_task + +# Bind @shared_task to the Redis-backed worker app +import app.worker # noqa: F401 +from app.db.session import SessionLocal +from app.models.models import Video +from app.services.media_analysis import ( + extract_frames, + detect_scene_cuts, + detect_beats, + generate_caption_embedding, + download_video_from_s3, +) + +logger = logging.getLogger(__name__) + + +@shared_task(name="tasks.media.analyze_media", max_retries=2, default_retry_delay=120) +def analyze_media(video_id: int): + """Download transcoded video, run frame extraction, scene detection, + beat tracking, and caption embedding, then write results to media_analysis.""" + + db = SessionLocal() + video = None + local_video_path = None + status = "processing" + + try: + video = db.query(Video).filter(Video.id == video_id).first() + if not video: + logger.error("analyze_media: video %s not found", video_id) + return + + # Ensure a media_analysis row exists + from app.models.models import Base + # Lazy import to avoid circular + from sqlalchemy import text + db.execute( + text( + "INSERT INTO media_analysis (video_id, status, created_at, updated_at) " + "VALUES (:vid, 'processing', now(), now()) " + "ON CONFLICT (video_id) DO UPDATE SET status='processing', updated_at=now()" + ), + {"vid": video_id}, + ) + db.commit() + + # Resolve the S3 key for the master HLS manifest or fallback mp4 + # The transcoded video lives under videos/{task_id}/master.m3u8 + # For analysis we want the MP4 source — but after finalization source is deleted. + # Use the cover/thumbnail as a proxy is not useful. + # Instead, we re-download from the first quality variant's segments. + # Actually: the original upload is deleted. We need the HLS stream. + # For frame extraction, we can use the HLS URL directly via ffmpeg. + video_s3_key = f"videos/{video.processing_key}/master.m3u8" if video.processing_key else None + + # Download the video — try HLS master first, fall back to direct mp4 + hls_url = video.video_url + if hls_url: + # ffmpeg can read HLS directly — download to a local mp4 for analysis + local_video_path = tempfile.mktemp(suffix=".mp4") + import subprocess + subprocess.run( + [ + "ffmpeg", "-i", hls_url, + "-c", "copy", + "-bsf:a", "aac_adtstoasc", + "-y", local_video_path, + ], + check=True, + capture_output=True, + timeout=900, + ) + logger.info("Downloaded HLS stream to %s for video %s", local_video_path, video_id) + else: + logger.error("analyze_media: no video_url for video %s", video_id) + status = "failed" + return + + # 1) Frame extraction + frame_paths = extract_frames(local_video_path, str(video_id)) + + # 2) Scene cuts + scene_cuts = detect_scene_cuts(local_video_path) + + # 3) Beat detection + beat_timestamps = detect_beats(local_video_path) + + # 4) Caption embedding + caption_embedding = generate_caption_embedding(video.title, video.tags) + + # Write results + from sqlalchemy import text as sa_text + db.execute( + sa_text( + "UPDATE media_analysis SET " + " status = 'done'," + " frame_sample_paths = :frames," + " scene_cuts = :scenes," + " beat_timestamps = :beats," + " caption_embedding = :embedding," + " updated_at = now() " + "WHERE video_id = :vid" + ), + { + "frames": frame_paths, + "scenes": scene_cuts, + "beats": beat_timestamps, + "embedding": str(caption_embedding) if caption_embedding else None, + "vid": video_id, + }, + ) + db.commit() + logger.info("Media analysis completed for video %s", video_id) + + except Exception as e: + logger.exception("Media analysis failed for video %s: %s", video_id, e) + status = "failed" + try: + if video: + from sqlalchemy import text as sa_text + db.execute( + sa_text( + "UPDATE media_analysis SET status='failed', updated_at=now() WHERE video_id = :vid" + ), + {"vid": video_id}, + ) + db.commit() + except Exception: + db.rollback() + finally: + if local_video_path and os.path.exists(local_video_path): + try: + os.unlink(local_video_path) + except OSError: + pass + db.close() diff --git a/backend/app/tasks/video_tasks.py b/backend/app/tasks/video_tasks.py index 4e64e875..52760aee 100644 --- a/backend/app/tasks/video_tasks.py +++ b/backend/app/tasks/video_tasks.py @@ -197,6 +197,13 @@ def _finalize_transcode_success( except Exception as e: logger.warning(f"Failed to delete source directory uploads/{task_id}/: {e}") + # Fire-and-forget: run media analysis pipeline asynchronously + try: + from app.tasks.media_tasks import analyze_media + analyze_media.delay(video_id) + except Exception as e: + logger.warning(f"Failed to enqueue media analysis for video {video_id}: {e}") + return {"status": "success", "video_id": video_id} diff --git a/backend/app/worker.py b/backend/app/worker.py index 2d73f71e..1321df9b 100644 --- a/backend/app/worker.py +++ b/backend/app/worker.py @@ -17,6 +17,8 @@ "app.tasks.video_tasks", "app.tasks.email_tasks", "app.tasks.morning_notifications", + "app.tasks.event_tasks", + "app.tasks.media_tasks", ] ) @@ -79,6 +81,11 @@ "task": "tasks.email.flush_social_batch", "schedule": crontab(hour=18, minute=0), }, + # Video engagement events — flush every 60 seconds + "flush-video-events": { + "task": "tasks.events.flush_video_events", + "schedule": 60.0, + }, }, ) diff --git a/backend/requirements.txt b/backend/requirements.txt index b9908ace..feaf06c4 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -29,3 +29,7 @@ fastapi-cache2[redis] tenacity stripe cryptography +scenedetect[opencv] +librosa +sentence-transformers +pgvector diff --git a/next-app/src/components/flash/FlashCard.jsx b/next-app/src/components/flash/FlashCard.jsx index e6514310..88368126 100644 --- a/next-app/src/components/flash/FlashCard.jsx +++ b/next-app/src/components/flash/FlashCard.jsx @@ -7,6 +7,7 @@ import { useRouter } from 'next/navigation'; import { viewVideo } from '@/lib/clientApi'; import { getStreamUrl, fetchStreamSignedUrl } from '@/lib/streamUrl'; import { useTrackHistory } from '@/hooks/useLibrary'; +import { useVideoEvents, generateSessionId } from '@/hooks/useVideoEvents'; import { useReport } from '@/context/ReportContext'; // Services @@ -53,6 +54,10 @@ const FlashCard = ({ // Interaction Tracking const entryTime = useRef(0); + // Engagement event tracking + const [eventSessionId] = useState(() => generateSessionId()); + const { fireView, fireSkip, checkThresholds, fireLike, fireShare } = useVideoEvents(video.id, null, eventSessionId); + // Smart Replay Config const isSmartMode = video.smart_replay || true; const smartStart = 0.25; @@ -213,6 +218,9 @@ const FlashCard = ({ const vDuration = videoRef.current.duration || video.duration || 0; trackingManager.startSession(video.id, vDuration); + // Engagement view event (replaces legacy viewVideo call) + fireView(); + viewTimer = setTimeout(async () => { try { await viewVideo(video.id); @@ -230,6 +238,9 @@ const FlashCard = ({ adaptiveDiscovery.recordWatch(video.id, watchMs, durTime * 1000, video.mood); trackingManager.endSession(video.id); + // Fire skip if user leaves before 25% + fireSkip(curTime); + if (curTime > 2) { trackHistory.mutate({ video_id: video.id, @@ -270,6 +281,9 @@ const FlashCard = ({ const startTime = dur * smartStart; videoRef.current.currentTime = startTime; } + + // Engagement thresholds + checkThresholds(curTime, dur); } } }; @@ -464,7 +478,7 @@ const FlashCard = ({ {/* Sidebar Actions */}
-
{ e.stopPropagation(); trackingManager.markLiked(video.id); onLike(video.id); }}> +
{ e.stopPropagation(); trackingManager.markLiked(video.id); fireLike(); onLike(video.id); }}>
@@ -481,7 +495,7 @@ const FlashCard = ({ {video.comments_count || 0}
-
{ e.stopPropagation(); onShare(video.id); }}> +
{ e.stopPropagation(); fireShare(); onShare(video.id); }}>
{video.shares || 0}
diff --git a/next-app/src/components/player/VideoPlayerV2.jsx b/next-app/src/components/player/VideoPlayerV2.jsx index e6360e52..0cfbfea0 100644 --- a/next-app/src/components/player/VideoPlayerV2.jsx +++ b/next-app/src/components/player/VideoPlayerV2.jsx @@ -10,6 +10,7 @@ import { useAuth } from '@/context/AuthContext'; import PreRollPlayer from '@/components/ads/PreRollPlayer'; import PauseOverlayAd from '@/components/ads/PauseOverlayAd'; import { useTrackHistory } from '@/hooks/useLibrary'; +import { useVideoEvents, generateSessionId } from '@/hooks/useVideoEvents'; // Resolution definitions: label, quality key for /stream-res, isPro gating // 'src' key means use the master HLS stream (default) @@ -57,6 +58,11 @@ const VideoPlayerV2 = ({ const sessionIdRef = useRef(null); const heartbeatIntervalRef = useRef(null); + // Engagement event tracking + const [eventSessionId] = useState(() => generateSessionId()); + const { fireView, fireSkip, checkThresholds, fireLike, fireShare } = useVideoEvents(videoId, user?.id, eventSessionId); + const viewFiredRef = useRef(false); + // UI State const [isPlaying, setIsPlaying] = useState(false); const [isMuted, setIsMuted] = useState(false); @@ -433,6 +439,12 @@ const VideoPlayerV2 = ({ return; } + // Fire engagement view event on first play + if (!viewFiredRef.current) { + viewFiredRef.current = true; + fireView(); + } + const startSession = async () => { if (!viewTicketRef.current) { try { @@ -486,6 +498,8 @@ const VideoPlayerV2 = ({ is_completed: curTime >= durTime * 0.9 && durTime > 0 }); } + // Fire skip if user leaves before 25% + fireSkip(curTime); } }; }, [videoId, token]); @@ -586,6 +600,9 @@ const VideoPlayerV2 = ({ const bufferedEnd = videoRef.current.buffered.end(videoRef.current.buffered.length - 1); setBufferedPercent((bufferedEnd / dur) * 100); } + + // Engagement thresholds + checkThresholds(cur, dur); }; const jump = (secs) => { diff --git a/next-app/src/components/watch/WatchView.jsx b/next-app/src/components/watch/WatchView.jsx index 3a84f73b..340b2c46 100644 --- a/next-app/src/components/watch/WatchView.jsx +++ b/next-app/src/components/watch/WatchView.jsx @@ -25,6 +25,7 @@ import VideoPreviewCard from '@/components/VideoPreviewCard'; import { useAuth } from '@/context/AuthContext'; import { useReport } from '@/context/ReportContext'; import { useWatchLaterToggle } from '@/hooks/useWatchLaterToggle'; +import { useVideoEvents, generateSessionId } from '@/hooks/useVideoEvents'; import { likeVideo, shareVideo, @@ -424,6 +425,10 @@ export default function WatchView({ const [showShareModal, setShowShareModal] = useState(false); const [navQueue, setNavQueue] = useState(() => relatedVideos || []); + // Engagement events — like/share from WatchView (view/watch tracked by VideoPlayerV2) + const [eventSessionId] = useState(() => generateSessionId()); + const { fireLike, fireShare } = useVideoEvents(video?.id, user?.id, eventSessionId); + const { isSaved: isSavedToWatchLater, isPending: watchLaterPending, @@ -535,6 +540,7 @@ export default function WatchView({ router.push('/login'); return; } + fireLike(); const wasLiked = !!video.liked_by_user; setVideo((prev) => ({ ...prev, @@ -562,6 +568,7 @@ export default function WatchView({ }; const handleShare = async () => { + fireShare(); setShowShareModal(true); try { await shareVideo(video.id); diff --git a/next-app/src/hooks/useVideoEvents.js b/next-app/src/hooks/useVideoEvents.js new file mode 100644 index 00000000..2b475b22 --- /dev/null +++ b/next-app/src/hooks/useVideoEvents.js @@ -0,0 +1,137 @@ +'use client'; + +import { useRef, useEffect, useCallback } from 'react'; +import { getClientApiBaseUrl } from '@/lib/streamUrl'; + +const FLUSH_INTERVAL_MS = 5000; + +let globalQueue = []; +let globalTimer = null; + +function getAuthHeaders() { + if (typeof window === 'undefined') return {}; + const token = localStorage.getItem('token'); + return token ? { Authorization: `Bearer ${token}` } : {}; +} + +async function flushQueue() { + if (globalQueue.length === 0) return; + const batch = globalQueue.splice(0, globalQueue.length); + try { + await fetch(`${getClientApiBaseUrl()}/events/video`, { + method: 'POST', + headers: { 'Content-Type': 'application/json', ...getAuthHeaders() }, + body: JSON.stringify({ events: batch }), + }); + } catch { + // Silent — events are best-effort + } +} + +function ensureTimer() { + if (globalTimer) return; + globalTimer = setInterval(() => { + flushQueue(); + }, FLUSH_INTERVAL_MS); + if (typeof window !== 'undefined') { + window.addEventListener('beforeunload', () => { + clearInterval(globalTimer); + globalTimer = null; + flushQueue(); + }); + } +} + +export function useVideoEvents(videoId, userId, sessionId) { + const firedRef = useRef({ + view: false, + watch_25: false, + watch_50: false, + watch_75: false, + complete: false, + }); + + const pushEvent = useCallback( + (eventType, watchSeconds) => { + if (!userId || !videoId) return; + globalQueue.push({ + user_id: userId, + video_id: videoId, + event_type: eventType, + watch_seconds: watchSeconds != null ? Math.floor(watchSeconds) : null, + session_id: sessionId || null, + }); + ensureTimer(); + }, + [videoId, userId, sessionId] + ); + + const fireOnce = useCallback( + (eventType, watchSeconds) => { + if (firedRef.current[eventType]) return; + firedRef.current[eventType] = true; + pushEvent(eventType, watchSeconds); + }, + [pushEvent] + ); + + const checkThresholds = useCallback( + (currentTime, duration) => { + if (!duration || duration <= 0) return; + const pct = currentTime / duration; + if (pct >= 0.25) fireOnce('watch_25', currentTime); + if (pct >= 0.50) fireOnce('watch_50', currentTime); + if (pct >= 0.75) fireOnce('watch_75', currentTime); + if (pct >= 0.95) fireOnce('complete', currentTime); + }, + [fireOnce] + ); + + const fireView = useCallback(() => { + fireOnce('view', 0); + }, [fireOnce]); + + const fireSkip = useCallback( + (watchSeconds) => { + if (firedRef.current.view && !firedRef.current.watch_25) { + fireOnce('skip', watchSeconds); + } + }, + [fireOnce] + ); + + const fireLike = useCallback(() => { + pushEvent('like', null); + }, [pushEvent]); + + const fireShare = useCallback(() => { + pushEvent('share', null); + }, [pushEvent]); + + const flush = useCallback(() => { + flushQueue(); + }, []); + + // Flush on unmount + fire skip if user leaves early + useEffect(() => { + return () => { + flush(); + }; + }, [flush]); + + return { + fireView, + fireSkip, + checkThresholds, + fireLike, + fireShare, + flush, + }; +} + +export function generateSessionId() { + if (typeof crypto !== 'undefined' && crypto.randomUUID) { + return crypto.randomUUID(); + } + return `${Date.now()}-${Math.random().toString(36).slice(2, 11)}`; +}