feat(update): add a Buzz price alerts panel to the notification bell

P5a of docs/plans/paid-model-price-tracking.md: one surface that answers "what
got cheaper / became free", without a permanent button (the grid filter was
dropped by owner decision, so the panel carries the actions itself).

- price_alert_since records when an alert started, so the panel can say
  "dropped 3 d ago" and count what is new since the user last looked; it is set
  on the first sight of an already-cheap version, preserved while the alert
  stands, and cleared when the price rises back above the threshold
- get_price_alerts() compares the threshold at read time (editing it takes
  effect immediately, no refresh needed) and returns both kinds in one list;
  model_type=None covers every type, which the shared update DB makes a single
  query. "became free" needs no price data, so it is reported even while price
  tracking is off
- GET /api/lm/price-alerts, registered once in MiscRoutes rather than per model
  type, decorating rows best-effort with the local model name and file path from
  the scanner indexes (a cold cache just omits them)
- a third tab in the notification bell: segments for under-threshold and
  became-free, the three states (tracking off / nothing matching / stale), and
  per-row actions (CivitAI always, Open when the model is local)
- two non-permanent entry points share one helper: the controls-bar updates
  dropdown and the global context menu, whose label carries the unread count
- unread state stays client-side (localStorage watermark); the count is fetched
  once on init and only when price tracking is enabled
- the per-type frontend client method is removed as dead code; the per-type
  backend route stays for the companion extension
This commit is contained in:
Will Miao
2026-10-04 15:08:48 +08:00
parent ad2402724b
commit 7ed19c185c
27 changed files with 2142 additions and 86 deletions
+161
View File
@@ -1723,3 +1723,164 @@ async def test_get_price_alerts_tolerates_bad_limit(tmp_path):
service = ModelUpdateService(str(tmp_path / "updates.sqlite"))
assert await service.get_price_alerts("lora", limit="not-a-number") == []
# --- Price alerts panel semantics (read-time threshold, across types) ---------
def _gated_response_for(version_id: int, ends_at: str) -> dict:
return {
"modelVersions": [
{
"id": version_id,
"baseModel": "Pony",
"availability": "Public",
"paidAccess": {"permanent": False, "endsAt": ends_at},
"files": [],
"images": [],
}
]
}
def _free_response_for(version_id: int) -> dict:
return {
"modelVersions": [
{
"id": version_id,
"baseModel": "Pony",
"availability": "Public",
"files": [],
"images": [],
}
]
}
@pytest.mark.asyncio
async def test_price_alerts_threshold_is_compared_at_read_time(tmp_path):
"""Changing the threshold must change panel membership without a refresh."""
service = _price_service(
tmp_path, price_tracking_enabled=True, price_alert_threshold_buzz=300
)
scanner = DummyScanner(LOCAL_RAW_DATA)
await service.refresh_for_model_type(
"lora", scanner, PriceProvider(GATED_RESPONSE, prices=_prices(250))
)
assert len(await service.get_price_alerts("lora")) == 1
# The stored alert state still says "hit"...
stored = await service.get_record("lora", 1)
assert stored.versions[0].price_alert_state is True
# ...but a tighter threshold excludes it immediately.
assert await service.get_price_alerts("lora", threshold_buzz=100) == []
assert len(await service.get_price_alerts("lora", threshold_buzz=300)) == 1
@pytest.mark.asyncio
async def test_price_alerts_span_all_model_types(tmp_path):
"""model_type=None is what the global panel uses: one list, all types."""
service = _price_service(
tmp_path, price_tracking_enabled=True, price_alert_threshold_buzz=1000
)
lora_scanner = DummyScanner([{"civitai": {"modelId": 1, "id": 11}}])
checkpoint_scanner = DummyScanner([{"civitai": {"modelId": 2, "id": 21}}])
await service.refresh_for_model_type(
"lora",
lora_scanner,
PriceProvider(_gated_response_for(12, "2999-01-01T00:00:00.000Z"), prices=_prices(250)),
)
await service.refresh_for_model_type(
"checkpoint",
checkpoint_scanner,
PriceProvider(
_gated_response_for(22, "2999-01-01T00:00:00.000Z"),
prices={22: {"price_buzz": 100}},
),
)
everything = await service.get_price_alerts()
assert {(a["modelType"], a["modelId"], a["versionId"]) for a in everything} == {
("lora", 1, 12),
("checkpoint", 2, 22),
}
# Cheapest first.
assert [a["priceBuzz"] for a in everything] == [100, 250]
assert [a["modelType"] for a in await service.get_price_alerts("checkpoint")] == [
"checkpoint"
]
@pytest.mark.asyncio
async def test_price_alert_since_tracks_the_crossing_and_clears(tmp_path):
service = _price_service(
tmp_path, price_tracking_enabled=True, price_alert_threshold_buzz=300
)
scanner = DummyScanner(LOCAL_RAW_DATA)
first = await service.refresh_for_model_type(
"lora",
scanner,
PriceProvider(_gated_response("2999-01-01T00:00:00.000Z"), prices=_prices(250)),
)
since = first[1].versions[0].price_alert_since
assert since is not None
# Still under the threshold: the original crossing time is preserved.
second = await service.refresh_for_model_type(
"lora",
scanner,
PriceProvider(_gated_response("2999-02-01T00:00:00.000Z"), prices=_prices(200)),
)
assert second[1].versions[0].price_alert_since == since
stored = await service.get_record("lora", 1)
assert stored.versions[0].price_alert_since == since
# Back above it: cleared, and the row leaves the panel.
third = await service.refresh_for_model_type(
"lora",
scanner,
PriceProvider(_gated_response("2999-03-01T00:00:00.000Z"), prices=_prices(900)),
)
assert third[1].versions[0].price_alert_since is None
assert await service.get_price_alerts("lora") == []
@pytest.mark.asyncio
async def test_price_alerts_report_became_free_without_price_tracking(tmp_path):
"""Gate transitions need no price data, so the "became free" half of the panel
works even while price tracking is switched off."""
service = _price_service(tmp_path) # tracking off
scanner = DummyScanner(LOCAL_RAW_DATA)
await service.refresh_for_model_type("lora", scanner, DummyProvider(GATED_RESPONSE))
assert await service.get_price_alerts("lora") == []
await service.refresh_for_model_type("lora", scanner, DummyProvider(FREE_RESPONSE))
alerts = await service.get_price_alerts("lora")
assert [alert["kind"] for alert in alerts] == ["became_free"]
assert alerts[0]["priceBuzz"] is None
assert alerts[0]["gateLapsedAt"] is not None
assert alerts[0]["versionId"] == 12
@pytest.mark.asyncio
async def test_newest_price_checked_at_reports_the_latest_fetch(tmp_path):
service = _price_service(tmp_path, price_tracking_enabled=True)
scanner = DummyScanner(LOCAL_RAW_DATA)
assert service.newest_price_checked_at() is None
await service.refresh_for_model_type(
"lora", scanner, PriceProvider(GATED_RESPONSE, prices=PRICE_PAYLOAD)
)
newest = service.newest_price_checked_at()
assert newest is not None
assert newest > 0