diff --git a/app/ffmpeg.py b/app/ffmpeg.py
index 7ed6990..50d9244 100644
--- a/app/ffmpeg.py
+++ b/app/ffmpeg.py
@@ -1,3 +1,4 @@
+import asyncio
 import shlex
 from pathlib import Path
 
@@ -99,3 +100,24 @@ def extract_output_path(args: list[str]) -> str | None:
             return arg
         i -= 1
     return None
+
+
+async def get_duration(input_path: str) -> float | None:
+    """Get duration of media file using ffprobe."""
+    try:
+        process = await asyncio.create_subprocess_exec(
+            "ffprobe",
+            "-v", "error",
+            "-show_entries", "format=duration",
+            "-of", "default=noprint_wrappers=1:nokey=1",
+            input_path,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        stdout, _ = await process.communicate()
+
+        if process.returncode == 0:
+            return float(stdout.decode().strip())
+    except (ValueError, OSError):
+        pass
+    return None
diff --git a/tests/test_ffmpeg.py b/tests/test_ffmpeg.py
index 73c81a0..2fcf4a7 100644
--- a/tests/test_ffmpeg.py
+++ b/tests/test_ffmpeg.py
@@ -1,6 +1,9 @@
 import pytest
+import asyncio
+from unittest.mock import AsyncMock, patch, MagicMock
 
 from app.ffmpeg import parse_command, resolve_paths, parse_progress, extract_output_path
+from app.models import Job
 
 
 def test_parse_simple_command():
@@ -99,3 +102,32 @@ def test_extract_output_path_complex():
     output_path = extract_output_path(args)
 
     assert output_path == "out.mp4"
+
+
+@pytest.mark.asyncio
+async def test_get_duration():
+    from app.ffmpeg import get_duration
+
+    # Mock ffprobe output
+    mock_process = MagicMock()
+    mock_process.communicate = AsyncMock(return_value=(b"120.5\n", b""))
+    mock_process.returncode = 0
+
+    with patch("asyncio.create_subprocess_exec", return_value=mock_process):
+        duration = await get_duration("/data/input.mp4")
+
+    assert duration == 120.5
+
+
+@pytest.mark.asyncio
+async def test_get_duration_failure():
+    from app.ffmpeg import get_duration
+
+    mock_process = MagicMock()
+    mock_process.communicate = AsyncMock(return_value=(b"", b"error"))
+    mock_process.returncode = 1
+
+    with patch("asyncio.create_subprocess_exec", return_value=mock_process):
+        duration = await get_duration("/data/input.mp4")
+
+    assert duration is None