diff --git a/backend/calendar_refresh.py b/backend/calendar_refresh.py new file mode 100644 index 0000000..dae11bc --- /dev/null +++ b/backend/calendar_refresh.py @@ -0,0 +1,108 @@ +from __future__ import annotations + +import asyncio +import logging +from collections.abc import Callable +from datetime import UTC, datetime, timedelta + +from fastapi import HTTPException +from sqlalchemy import or_, select +from sqlalchemy.ext.asyncio import AsyncSession + +from . import calendar as calendar_service +from .models import CalendarSubscription, utcnow + +logger = logging.getLogger(__name__) +DEFAULT_REFRESH_INTERVAL = timedelta(minutes=15) +DEFAULT_POLL_SECONDS = 60 + + +async def refresh_subscription_cache( + db: AsyncSession, + row: CalendarSubscription, + *, + fail_without_cache: bool = True, +) -> bool: + requested_url = row.url + requested_version = getattr(row, "updated_at", None) + try: + result = await asyncio.to_thread( + calendar_service.fetch_calendar, + requested_url, + etag=row.etag, + last_modified=row.last_modified, + ) + await db.refresh(row) + if row.url != requested_url or getattr(row, "updated_at", None) != requested_version: + return False + if result.not_modified: + if not row.ics_cache: + raise HTTPException(502, "calendar returned not modified without cache") + elif result.content is not None: + # Parse before replacing a known-good cache. + calendar_service.parse_ics_events( + result.content, + row.name, + row.color, + datetime.now(UTC) - timedelta(days=1), + datetime.now(UTC) + timedelta(days=1), + "UTC", + ) + row.ics_cache = result.content.decode("utf-8-sig") + row.etag = result.etag + row.last_modified = result.last_modified + row.refreshed_at = utcnow() + row.last_error = None + await db.commit() + return True + except Exception as exc: + error = exc.detail if isinstance(exc, HTTPException) else str(exc) + if not row.ics_cache and fail_without_cache: + await db.rollback() + raise HTTPException(502, error) from exc + row.last_error = error + await db.commit() + return False + + +async def refresh_due_subscriptions( + db: AsyncSession, + *, + now: datetime | None = None, + refresh_interval: timedelta = DEFAULT_REFRESH_INTERVAL, +) -> int: + current = now or datetime.now(UTC) + cutoff = current - refresh_interval + rows = (await db.scalars(select(CalendarSubscription).where( + CalendarSubscription.enabled.is_(True), + or_( + CalendarSubscription.refreshed_at.is_(None), + CalendarSubscription.refreshed_at <= cutoff, + ), + ).order_by(CalendarSubscription.refreshed_at, CalendarSubscription.created_at))).all() + for row in rows: + try: + await refresh_subscription_cache(db, row, fail_without_cache=False) + except Exception: + await db.rollback() + logger.exception("Unexpected calendar refresh failure", extra={"subscription_id": str(row.id)}) + return len(rows) + + +async def calendar_refresh_loop( + session_factory: Callable[[], AsyncSession], + stop_event: asyncio.Event, + *, + refresh_interval: timedelta = DEFAULT_REFRESH_INTERVAL, + poll_seconds: int = DEFAULT_POLL_SECONDS, +) -> None: + while not stop_event.is_set(): + try: + async with session_factory() as db: + await refresh_due_subscriptions(db, refresh_interval=refresh_interval) + except Exception: + logger.exception("Calendar background refresh cycle failed") + try: + await asyncio.wait_for(stop_event.wait(), timeout=poll_seconds) + except TimeoutError: + pass diff --git a/backend/calendar_router.py b/backend/calendar_router.py index 49e17e5..6946c4c 100644 --- a/backend/calendar_router.py +++ b/backend/calendar_router.py @@ -1,4 +1,4 @@ -from datetime import UTC, datetime, timedelta +from datetime import datetime, timedelta from uuid import UUID from zoneinfo import ZoneInfo, ZoneInfoNotFoundError @@ -9,8 +9,9 @@ from sqlalchemy.ext.asyncio import AsyncSession from . import calendar as calendar_service from .auth import current_user +from .calendar_refresh import refresh_subscription_cache from .db import get_db -from .models import CalendarSubscription, User, utcnow +from .models import CalendarSubscription, User router = APIRouter(prefix="/api/v1", tags=["calendar"]) MAX_WINDOW = timedelta(days=366) @@ -90,37 +91,6 @@ async def _owned(db: AsyncSession, user_id: UUID, subscription_id: UUID) -> Cale return row -async def _refresh(db: AsyncSession, row: CalendarSubscription) -> None: - try: - result = await __import__("asyncio").to_thread( - calendar_service.fetch_calendar, row.url, etag=row.etag, last_modified=row.last_modified - ) - if result.not_modified: - if not row.ics_cache: - raise HTTPException(502, "calendar returned not modified without cache") - elif result.content is not None: - # Parse before replacing a known-good cache. - calendar_service.parse_ics_events( - result.content, - row.name, - row.color, - datetime.now(UTC) - timedelta(days=1), - datetime.now(UTC) + timedelta(days=1), - "UTC", - ) - row.ics_cache = result.content.decode("utf-8-sig") - row.etag = result.etag - row.last_modified = result.last_modified - row.refreshed_at = utcnow() - row.last_error = None - except Exception as exc: - row.last_error = exc.detail if isinstance(exc, HTTPException) else str(exc) - if not row.ics_cache: - await db.rollback() - raise HTTPException(502, row.last_error) from exc - await db.commit() - - @router.get("/calendar-subscriptions", response_model=list[SubscriptionOut]) async def list_subscriptions(user: User = Depends(current_user), db: AsyncSession = Depends(get_db)): rows = (await db.scalars(select(CalendarSubscription).where( @@ -139,7 +109,7 @@ async def create_subscription( row = CalendarSubscription(user_id=user.id, **payload.model_dump()) db.add(row) await db.flush() - await _refresh(db, row) + await refresh_subscription_cache(db, row) await db.refresh(row) return _out(row) @@ -183,7 +153,7 @@ async def refresh_subscription( db: AsyncSession = Depends(get_db), ): row = await _owned(db, user.id, subscription_id) - await _refresh(db, row) + await refresh_subscription_cache(db, row) await db.refresh(row) return _out(row) @@ -209,7 +179,7 @@ async def calendar_events( sources = [] for row in rows: if not row.ics_cache: - await _refresh(db, row) + await refresh_subscription_cache(db, row) try: parsed = calendar_service.parse_ics_events( row.ics_cache or "", row.name, row.color, start, end, user.timezone, diff --git a/backend/db.py b/backend/db.py index 18a1c1c..6da4cca 100644 --- a/backend/db.py +++ b/backend/db.py @@ -22,6 +22,12 @@ def get_engine(): return _engine +def get_session_factory() -> async_sessionmaker[AsyncSession]: + get_engine() + assert _session_factory is not None + return _session_factory + + def reset_engine() -> None: global _engine, _session_factory _engine = None @@ -29,9 +35,7 @@ def reset_engine() -> None: async def get_db() -> AsyncIterator[AsyncSession]: - get_engine() - assert _session_factory is not None - async with _session_factory() as session: + async with get_session_factory()() as session: yield session diff --git a/backend/main.py b/backend/main.py index 2b64453..6772eb7 100644 --- a/backend/main.py +++ b/backend/main.py @@ -1,3 +1,4 @@ +import asyncio import base64 import json import logging @@ -30,8 +31,9 @@ from .auth import ( verify_password, ) from .backup import router as backup_router +from .calendar_refresh import calendar_refresh_loop from .calendar_router import router as calendar_router -from .db import create_schema, get_db +from .db import create_schema, get_db, get_session_factory from .models import ( AppState, Attachment, @@ -80,7 +82,19 @@ async def lifespan(app: FastAPI): if get_settings().auto_create_schema: await create_schema() - yield + stop_calendar_refresh = asyncio.Event() + calendar_refresh_task = asyncio.create_task( + calendar_refresh_loop(get_session_factory(), stop_calendar_refresh) + ) + try: + yield + finally: + stop_calendar_refresh.set() + calendar_refresh_task.cancel() + try: + await calendar_refresh_task + except asyncio.CancelledError: + pass app = FastAPI( diff --git a/frontend/src/CalendarPanel.test.ts b/frontend/src/CalendarPanel.test.ts index b334d9c..0ed791f 100644 --- a/frontend/src/CalendarPanel.test.ts +++ b/frontend/src/CalendarPanel.test.ts @@ -92,6 +92,7 @@ describe('CalendarPanel',()=>{ vi.useRealTimers() }) it('keeps the newest week response when requests finish out of order',async()=>{ + vi.useFakeTimers();vi.setSystemTime(new Date(2026,8,20,10)) const pending:Array<{url:string;resolve:(response:Response)=>void}>=[] const fetchMock=vi.fn((url:string)=>String(url).includes('calendar-events')?new Promise(resolve=>pending.push({url:String(url),resolve})):Promise.resolve(json(subscriptions))) const {host}=await mount(fetchMock) @@ -102,6 +103,7 @@ describe('CalendarPanel',()=>{ pending[0].resolve(json({events:[{...events[0],id:'old',title:'旧一周'}],sources:[]}));await flush() expect(host.textContent).toContain('新一周') expect(host.textContent).not.toContain('旧一周') + vi.useRealTimers() }) it('supports week navigation and today',async()=>{ const fetchMock=vi.fn((url:string)=>Promise.resolve(json(url.includes('calendar-events')?{events:[],sources:[]}:subscriptions))) diff --git a/tests/test_calendar_subscriptions.py b/tests/test_calendar_subscriptions.py index 1984f77..d1e1508 100644 --- a/tests/test_calendar_subscriptions.py +++ b/tests/test_calendar_subscriptions.py @@ -1,5 +1,7 @@ +import asyncio import socket -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace import pytest from fastapi import HTTPException @@ -9,6 +11,7 @@ from backend.calendar import ( parse_ics_events, validate_calendar_url, ) +from backend.calendar_refresh import refresh_due_subscriptions, refresh_subscription_cache ICS = b"""BEGIN:VCALENDAR\r VERSION:2.0\r @@ -234,3 +237,117 @@ def test_subscription_ownership_is_strict(client, monkeypatch): client.post("/api/v1/auth/logout") assert client.patch(f"/api/v1/calendar-subscriptions/{sub['id']}", json={"name": "x"}).status_code == 401 assert client.delete(f"/api/v1/calendar-subscriptions/{sub['id']}").status_code == 401 + + +def test_due_enabled_subscriptions_refresh_automatically(client, monkeypatch): + client = initialized(client) + monkeypatch.setattr( + "backend.calendar.validate_calendar_url", + lambda url: (url, "93.184.216.34", 443), + ) + calls = [] + + def fetch(url, *, etag=None, last_modified=None): + calls.append(url) + return FetchResult(ICS, None, None, False) + + monkeypatch.setattr("backend.calendar.fetch_calendar", fetch) + client.post( + "/api/v1/calendar-subscriptions", + json={"name": "enabled", "url": "https://example.com/enabled.ics"}, + ) + client.post( + "/api/v1/calendar-subscriptions", + json={"name": "disabled", "url": "https://example.com/disabled.ics", "enabled": False}, + ) + + async def refresh(): + from sqlalchemy.ext.asyncio import async_sessionmaker + + from backend.db import get_engine + + factory = async_sessionmaker(get_engine(), expire_on_commit=False) + async with factory() as db: + return await refresh_due_subscriptions( + db, + now=datetime.now(UTC) + timedelta(minutes=16), + refresh_interval=timedelta(minutes=15), + ) + + assert asyncio.run(refresh()) == 1 + assert calls == [ + "https://example.com/enabled.ics", + "https://example.com/disabled.ics", + "https://example.com/enabled.ics", + ] + + +def test_failed_cached_refresh_preserves_last_success_timestamp(monkeypatch): + original_refresh = datetime(2026, 9, 20, tzinfo=UTC) + row = SimpleNamespace( + id="source-1", + url="https://example.com/work.ics", + name="work", + color="#123456", + ics_cache=ICS.decode(), + etag=None, + last_modified=None, + refreshed_at=original_refresh, + last_error=None, + ) + + class FakeDb: + async def refresh(self, _row): + pass + + async def commit(self): + pass + + async def rollback(self): + pass + + def fail(*args, **kwargs): + raise HTTPException(502, "upstream down") + + monkeypatch.setattr("backend.calendar.fetch_calendar", fail) + assert asyncio.run(refresh_subscription_cache(FakeDb(), row)) is False + assert row.refreshed_at == original_refresh + assert row.last_error == "upstream down" + + +def test_refresh_discards_response_when_url_changes_in_flight(monkeypatch): + row = SimpleNamespace( + id="source-1", + url="https://example.com/old.ics", + name="work", + color="#123456", + ics_cache="old cache", + etag=None, + last_modified=None, + refreshed_at=datetime(2026, 9, 20, tzinfo=UTC), + last_error=None, + ) + + class FakeDb: + committed = False + + async def refresh(self, target): + target.url = "https://example.com/new.ics" + target.ics_cache = None + target.refreshed_at = None + + async def commit(self): + self.committed = True + + async def rollback(self): + pass + + monkeypatch.setattr( + "backend.calendar.fetch_calendar", + lambda *args, **kwargs: FetchResult(ICS, None, None, False), + ) + db = FakeDb() + assert asyncio.run(refresh_subscription_cache(db, row)) is False + assert row.url == "https://example.com/new.ics" + assert row.ics_cache is None + assert db.committed is False