from __future__ import annotations import hashlib import time from email.utils import parsedate_to_datetime def build_etag(file_id: str, file_data: dict, download: bool = False) -> str: """Strong validator for a stored file. The blob behind a file_id never changes (updateFile only patches metadata), so uploaded_at plus the storage id identify the bytes. The download flag is part of the tag because it changes Content-Disposition. """ parts = [ str(file_id), str(file_data.get('uploaded_at', '')), str(file_data.get('db_image_url', '')), str(file_data.get('file_name', '')), 'attachment' if download else 'inline', ] digest = hashlib.sha256('\x1f'.join(parts).encode('utf-8')).hexdigest()[:32] return f'"{digest}"' def _split_etags(header: str) -> list[str]: return [value.strip() for value in header.split(',') if value.strip()] def etag_matches(if_none_match: str | None, etag: str) -> bool: """RFC 9110 If-None-Match: '*' matches, weak comparison otherwise.""" if not if_none_match or not etag: return False header = if_none_match.strip() if header == '*': return True def normalize(value: str) -> str: return value.removeprefix('W/') target = normalize(etag) return any(normalize(candidate) == target for candidate in _split_etags(header)) def _parse_http_date(value: str): try: return parsedate_to_datetime(value) except (TypeError, ValueError): return None def not_modified_since(if_modified_since: str | None, last_modified_epoch: float | None) -> bool: if not if_modified_since or last_modified_epoch is None: return False parsed = _parse_http_date(if_modified_since) if parsed is None: return False # HTTP dates have second precision, so compare truncated values. return int(last_modified_epoch) <= int(parsed.timestamp()) def is_fresh( if_none_match: str | None, if_modified_since: str | None, etag: str, last_modified_epoch: float | None, ) -> bool: """Decide a 304 from metadata alone, before any blob is fetched.""" if if_none_match is not None: # An ETag was offered, so it alone decides and the date is ignored. return etag_matches(if_none_match, etag) return not_modified_since(if_modified_since, last_modified_epoch) def remaining_seconds(expires_at, now_ms: float | None = None) -> int | None: """Seconds until expires_at (epoch ms), or None when the file never expires.""" if expires_at in (None, '', 0): return None try: expires_ms = float(expires_at) except (TypeError, ValueError): return None if now_ms is None: now_ms = time.time() * 1000 return int((expires_ms - now_ms) / 1000) def _clamped_max_age(expires_at, max_age: int) -> int: if max_age <= 0: return 0 remaining = remaining_seconds(expires_at) if remaining is None: return max_age return min(max_age, max(0, remaining)) def cache_control(expires_at, max_age: int) -> str: """Browser caching for immutable blobs, clamped to the remaining lifetime.""" clamped = _clamped_max_age(expires_at, max_age) return f'public, max-age={clamped}' if clamped > 0 else 'no-cache' def edge_cache_control(expires_at, max_age: int) -> str: """Separate, deliberately short TTL for Cloudflare. Shared caches keep serving a file after it was deleted or expired at the origin, so the edge gets a much smaller window than the browser does. """ clamped = _clamped_max_age(expires_at, max_age) return f'public, max-age={clamped}' if clamped > 0 else 'no-store'