feat: auto-refresh calendar subscriptions
This commit is contained in:
@@ -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
|
||||
@@ -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,
|
||||
|
||||
+7
-3
@@ -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
|
||||
|
||||
|
||||
|
||||
+16
-2
@@ -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(
|
||||
|
||||
@@ -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<Response>(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)))
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user