mirror of
https://github.com/HyPnoTiiK/stream-fusion.git
synced 2026-07-28 15:02:46 +00:00
0c1d305784
Add a two-level caching architecture for debrid service availability lookups: - L1: Redis (existing) with 10-day TTL for fast in-memory access - L2: PostgreSQL with 30-day TTL for persistent caching across Redis restarts Changes: - Add `debrid_cache` table with unique constraint on (info_hash, service) and indexes for efficient queries - Create `DebridCacheDAO` with batch get/upsert and invalidation methods - Add pgCron job to periodically clean up expired cache entries - Extend `get_availability_bulk_cached` to check PG on Redis miss, warm Redis from PG hits, and write confirmed-cached results to PG - Add `invalidate_availability_cache` method to remove false positives from both caches when NO_CACHE_VIDEO_URL is returned during playback - Integrate `db_session` dependency into playback and search views - Add retry handling to Yggflix API (3 retries, ignore Retry-After headers) and rate-limiting delay (0.3s) between consecutive queries
105 lines
No EOL
3.9 KiB
Python
105 lines
No EOL
3.9 KiB
Python
from sqlalchemy import text
|
|
from sqlalchemy.engine import make_url
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
|
|
from stream_fusion.logging_config import logger
|
|
from stream_fusion.settings import settings
|
|
|
|
|
|
async def create_database() -> None:
|
|
"""Create a database."""
|
|
db_url = make_url(str(settings.pg_url.with_path("/postgres")))
|
|
engine = create_async_engine(db_url, isolation_level="AUTOCOMMIT")
|
|
|
|
async with engine.connect() as conn:
|
|
database_existance = await conn.execute(
|
|
text(
|
|
f"SELECT 1 FROM pg_database WHERE datname='{settings.pg_base}'", # noqa: S608
|
|
),
|
|
)
|
|
database_exists = database_existance.scalar() == 1
|
|
|
|
if database_exists:
|
|
await drop_database()
|
|
|
|
async with engine.connect() as conn:
|
|
await conn.execute(
|
|
text(
|
|
f'CREATE DATABASE "{settings.pg_base}" ENCODING "utf8" TEMPLATE template1', # noqa: E501
|
|
),
|
|
)
|
|
|
|
|
|
async def drop_database() -> None:
|
|
"""Drop current database."""
|
|
db_url = make_url(str(settings.pg_url.with_path("/postgres")))
|
|
engine = create_async_engine(db_url, isolation_level="AUTOCOMMIT")
|
|
async with engine.connect() as conn:
|
|
disc_users = (
|
|
"SELECT pg_terminate_backend(pg_stat_activity.pid) " # noqa: S608
|
|
"FROM pg_stat_activity "
|
|
f"WHERE pg_stat_activity.datname = '{settings.pg_base}' "
|
|
"AND pid <> pg_backend_pid();"
|
|
)
|
|
await conn.execute(text(disc_users))
|
|
await conn.execute(text(f'DROP DATABASE "{settings.pg_base}"'))
|
|
|
|
|
|
async def init_db_cleanup_function(engine):
|
|
async with engine.begin() as conn:
|
|
result = await conn.execute(text(
|
|
"SELECT 1 FROM pg_proc WHERE proname = 'delete_expired_api_keys'"
|
|
))
|
|
function_exists = result.scalar() is not None
|
|
|
|
if not function_exists:
|
|
await conn.execute(text("""
|
|
CREATE OR REPLACE FUNCTION delete_expired_api_keys() RETURNS void AS $$
|
|
BEGIN
|
|
DELETE FROM api_keys
|
|
WHERE NOT never_expire
|
|
AND expiration_date < NOW() - INTERVAL '7 days'
|
|
AND (latest_query_date IS NULL OR latest_query_date < NOW() - INTERVAL '7 days');
|
|
RAISE NOTICE 'Deleted % expired API keys', ROW_COUNT;
|
|
END;
|
|
$$ LANGUAGE plpgsql;
|
|
"""))
|
|
|
|
await conn.execute(text("CREATE EXTENSION IF NOT EXISTS pg_cron;"))
|
|
|
|
await conn.execute(text("""
|
|
SELECT cron.schedule('0 */6 * * *', $$SELECT delete_expired_api_keys()$$);
|
|
"""))
|
|
|
|
logger.info("API key cleanup function created and scheduled.")
|
|
else:
|
|
logger.info("API key cleanup function already exists.")
|
|
|
|
# Debrid cache cleanup — delete entries whose expires_at has passed
|
|
result = await conn.execute(text(
|
|
"SELECT 1 FROM pg_proc WHERE proname = 'delete_expired_debrid_cache'"
|
|
))
|
|
debrid_cleanup_exists = result.scalar() is not None
|
|
|
|
if not debrid_cleanup_exists:
|
|
await conn.execute(text("""
|
|
CREATE OR REPLACE FUNCTION delete_expired_debrid_cache() RETURNS void AS $$
|
|
DECLARE deleted_count INTEGER;
|
|
BEGIN
|
|
DELETE FROM debrid_cache
|
|
WHERE expires_at < EXTRACT(EPOCH FROM NOW())::BIGINT;
|
|
GET DIAGNOSTICS deleted_count = ROW_COUNT;
|
|
IF deleted_count > 0 THEN
|
|
RAISE NOTICE 'Deleted % expired debrid cache entries', deleted_count;
|
|
END IF;
|
|
END;
|
|
$$ LANGUAGE plpgsql;
|
|
"""))
|
|
|
|
await conn.execute(text("""
|
|
SELECT cron.schedule('30 */6 * * *', $$SELECT delete_expired_debrid_cache()$$);
|
|
"""))
|
|
|
|
logger.info("Debrid cache cleanup function created and scheduled.")
|
|
else:
|
|
logger.info("Debrid cache cleanup function already exists.") |