Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 43 additions & 0 deletions .github/workflows/offline-media-tests.yml
Original file line number Diff line number Diff line change
@@ -0,0 +1,43 @@
name: Offline Media Tests

on:
pull_request:
paths:
- 'backend/**'
- 'superbrain-app/**'
- '.github/workflows/offline-media-tests.yml'
push:
branches: [experimental/video-playback]
workflow_dispatch:

permissions:
contents: read

jobs:
backend:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install HTTP test dependencies
run: python -m pip install fastapi httpx python-multipart requests
- name: Test media storage, endpoint and sync
run: python -m unittest backend.tests.test_media_metadata backend.tests.test_media_retention backend.tests.test_media_endpoint backend.tests.test_instagram_temp_cleanup backend.tests.test_sync_pagination -v

client:
runs-on: ubuntu-latest
defaults:
run:
working-directory: superbrain-app
steps:
- uses: actions/checkout@v4
- uses: actions/setup-node@v5
with:
node-version: '22'
cache: npm
cache-dependency-path: superbrain-app/package-lock.json
- run: npm ci
- run: npm test
- run: npm run typecheck
4 changes: 4 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,10 @@ Everything is stored in a local SQLite database **you own** — no cloud subscri

## Features

Offline video playback includes a server cache with 30-day retention and hourly
cleanup. See [offline media storage](docs/OFFLINE_MEDIA.md) for capacity estimates,
configuration, and tests before deploying it.

### ✨ Content Analysis

| Feature | Description |
Expand Down
1 change: 1 addition & 0 deletions backend/.dockerignore
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ config/localtunnel.log
# Logs and transient artifacts
logs/
temp/
media/
error_log_utf8.txt
failures.txt
final_test.txt
Expand Down
7 changes: 7 additions & 0 deletions backend/.env.example
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,13 @@ API_PORT=5000
DATABASE_PATH=superbrain.db
DATABASE_TIMEOUT=30

# ──── Server Offline Video Cache ────
# Age is measured from save, not playback. See docs/OFFLINE_MEDIA.md.
# Export these variables for local runs; Docker Compose reads .env directly.
MEDIA_RETENTION_DAYS=30
MEDIA_CLEANUP_INTERVAL_SECONDS=3600
# MEDIA_PATH defaults to backend/media (Compose uses /app/media).

# ──── AI Provider API Keys ────
# Get these from the respective provider accounts
GROQ_API_KEY=
Expand Down
52 changes: 45 additions & 7 deletions backend/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,13 +22,15 @@
import secrets
import string
import threading
import time
from pathlib import Path
import time
from pathlib import Path
from contextlib import asynccontextmanager

# Import database module
from core.database import get_db
from core.link_checker import validate_link
from core.media_store import get_media_dir
from core.media_retention import positive_env_int, sweep_media

# Configure logging
logging.basicConfig(
Expand Down Expand Up @@ -87,11 +89,42 @@ async def verify_token(request: Request, x_api_key: str = Header(None, descripti
)
return actual_token

# Initialize FastAPI app
app = FastAPI(
async def run_media_cleanup():
"""Run disk/DB maintenance off the event loop; retry failures next interval."""
try:
await asyncio.to_thread(sweep_media, get_media_dir(), db.db_path)
except Exception:
logger.exception("Media cache cleanup failed; retrying at the next interval")


async def media_cleanup_loop(interval: int):
while True:
await asyncio.sleep(interval)
await run_media_cleanup()


@asynccontextmanager
async def lifespan(app: FastAPI):
await run_media_cleanup()
cleanup_task = asyncio.create_task(media_cleanup_loop(
positive_env_int("MEDIA_CLEANUP_INTERVAL_SECONDS", 3600)
))
try:
yield
finally:
cleanup_task.cancel()
try:
await cleanup_task
except asyncio.CancelledError:
pass


# Initialize FastAPI app
app = FastAPI(
title="SuperBrain",
description="AI-powered Instagram content analysis with caching",
version="1.02"
version="1.02",
lifespan=lifespan,
)

# CORS configuration
Expand Down Expand Up @@ -144,7 +177,12 @@ async def add_security_headers(request: Request, call_next):

def _resolve_media_file(filename: str) -> Path:
"""Resolve a media filename safely inside backend/media."""
if Path(filename).name != filename or not filename.lower().endswith(".mp4"):
if (
not filename
or any(char in filename for char in ("/", "\\", "\x00"))
or Path(filename).name != filename
or not filename.lower().endswith(".mp4")
):
raise HTTPException(status_code=400, detail="Invalid media filename")

media_root = _MEDIA_DIR.resolve()
Expand All @@ -160,7 +198,7 @@ def _resolve_media_file(filename: str) -> Path:
return media_path


@app.get("/api/v1/media/{filename}")
@app.get("/api/v1/media/{filename:path}")
async def get_media_file(filename: str, token: str = Depends(verify_token)):
"""
Stream a downloaded offline media file.
Expand Down
138 changes: 138 additions & 0 deletions backend/core/media_retention.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,138 @@
"""Retention for the server's disposable offline-video cache.

SQLite write locks coordinate publishers and sweepers across API workers and
analysis subprocesses, including on Windows. They are released after a crash.
"""

from contextlib import contextmanager
from datetime import datetime, timezone
import logging
import os
from pathlib import Path
import shutil
import sqlite3
import stat
import time

logger = logging.getLogger(__name__)
DAY_SECONDS = 86400
ORPHAN_GRACE_SECONDS = 3600


def positive_env_int(name: str, default: int) -> int:
try:
value = int(os.getenv(name, str(default)))
if value > 0:
return value
except ValueError:
pass
logger.warning("Invalid %s; using %s", name, default)
return default


@contextmanager
def media_lock(path: Path, timeout: float = 30):
"""Use a small, dedicated SQLite file as a process-safe exclusive lock."""
connection = sqlite3.connect(str(path), timeout=timeout)
try:
connection.execute("BEGIN IMMEDIATE")
yield
finally:
connection.rollback()
connection.close()


def sweep_media(media_dir: Path, database_path: Path, *, now: float | None = None,
retention_days: int | None = None) -> dict[str, int]:
"""Expire MP4s, unreferenced files, and abandoned download workspaces.

Use a separate DB connection so the sweep never shares the API's connection
across threads. A failed DB query must never be mistaken for zero references.
A one-hour grace protects the gap between publishing a file and saving its
analysis row. Only our MP4s/download artifacts are managed, never symlinks.
"""
now = time.time() if now is None else now
retention_days = (positive_env_int("MEDIA_RETENTION_DAYS", 30)
if retention_days is None else retention_days)
if retention_days <= 0:
raise ValueError("retention_days must be positive")
result = {"deleted_files": 0, "deleted_downloads": 0, "bytes_freed": 0,
"cleared_rows": 0, "failed_files": 0}
if not media_dir.exists():
return result

with media_lock(media_dir / ".maintenance.sqlite3"):
# mode=rw avoids creating a blank DB if DATABASE_PATH is incorrect.
connection = sqlite3.connect(
database_path.resolve().as_uri() + "?mode=rw", uri=True, timeout=30,
)
try:
# Serialize with save/delete/restore while inspecting references.
connection.execute("BEGIN IMMEDIATE")
rows = connection.execute(
"SELECT local_filename, is_hidden FROM analyses "
"WHERE local_filename IS NOT NULL AND local_filename != ''"
).fetchall()
active_names = {name for name, hidden in rows if not hidden}

for path in media_dir.iterdir():
try:
info = path.lstat()
if stat.S_ISLNK(info.st_mode):
continue
age = now - info.st_mtime
if stat.S_ISDIR(info.st_mode):
if path.name.startswith(".download-") and age >= ORPHAN_GRACE_SECONDS:
try:
# A running download holds this lease even if its
# directory timestamp is old. Never remove it.
with media_lock(path / ".active.sqlite3", timeout=0):
pass
except sqlite3.OperationalError:
continue
shutil.rmtree(path)
result["deleted_downloads"] += 1
continue
if not stat.S_ISREG(info.st_mode):
continue
if path.suffix.lower() == ".mp4":
expired = age >= retention_days * DAY_SECONDS
orphaned = path.name not in active_names and age >= ORPHAN_GRACE_SECONDS
if not (expired or orphaned):
continue
elif path.suffix.lower() in {".part", ".ytdl", ".m4a", ".webm"}:
# Leftovers from versions that downloaded directly here.
if age < ORPHAN_GRACE_SECONDS:
continue
else:
continue
path.unlink()
result["deleted_files"] += 1
result["bytes_freed"] += info.st_size
except OSError:
result["failed_files"] += 1
logger.exception("Could not remove cached media %s", path.name)

# Repair evictions and files removed externally/on a prior crash.
# Failed unlinks retain their metadata and are retried next sweep.
# Keep the database's existing naive-UTC ISO representation.
timestamp = (
datetime.fromtimestamp(now, timezone.utc)
.replace(tzinfo=None)
.isoformat()
)
for name in {name for name, _ in rows}:
if Path(name).name != name or "/" in name or "\\" in name or "\x00" in name:
continue
path = media_dir / name
if not path.exists():
cursor = connection.execute(
"UPDATE analyses SET local_filename = '', media_file_size = 0, "
"updated_at = ? WHERE local_filename = ?", (timestamp, name),
)
result["cleared_rows"] += cursor.rowcount
connection.commit()
finally:
connection.close()
logger.info("Media cache sweep: %s", result)
return result
Loading