alexei.dolgolyov/notify-bridge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 31 Wiki Activity

feat: add Scheduler provider + multi-provider UX fixes

Browse Source 
Scheduler provider:
- Virtual provider (no external service) that emits SCHEDULED_MESSAGE
  events on user-defined intervals or cron expressions
- Custom variables stored in tracker filters, flattened into template context
- fire_count persists across triggers via tracker state
- APScheduler CronTrigger support for cron-mode schedules
- Default templates (EN+RU), seeded on startup

Multi-provider UX fixes:
- Tracking config hides Immich-specific sections (periodic, scheduled,
  memory, asset display) for non-Immich providers
- Command config driven by provider capabilities — hides commands/settings
  for providers without bot commands
- Template config hides empty "Scheduled Messages" group
- Test menu on tracker targets is provider-aware (Immich shows all 4 test
  types, others show only basic)
- Removed redundant Test button from tracker card
- System-owned tracking configs (user_id=0) seeded for Gitea + Scheduler
- Fixed ownership checks to allow system configs in tracker-target links
- Capabilities cache shared across template-configs and command-configs
- Command tracker bot selector uses EntitySelect instead of raw select
- Sample context includes Gitea + Scheduler variables for template preview
This commit is contained in:
Admin
2026-03-22 15:50:51 +03:00
parent 6d28cfb8d8
commit 0562f78b35
30 changed files with 688 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/notify_bridge_server/api/notification_tracker_targets.py
+2 -2
View File
@@ -94,7 +94,7 @@ async def create_notification_tracker_target(
# Validate config ownership + provider type match
if body.tracking_config_id:
tc = await session.get(TrackingConfig, body.tracking_config_id)
if not tc or tc.user_id != user.id:
if not tc or (tc.user_id != user.id and tc.user_id != 0):
raise HTTPException(status_code=404, detail="Tracking config not found")
if tc.provider_type != provider.type:
raise HTTPException(
@@ -139,7 +139,7 @@ async def update_notification_tracker_target(
# Validate config ownership + provider type match if being changed
if "tracking_config_id" in updates and updates["tracking_config_id"]:
tc = await session.get(TrackingConfig, updates["tracking_config_id"])
if not tc or tc.user_id != user.id:
if not tc or (tc.user_id != user.id and tc.user_id != 0):
raise HTTPException(status_code=404, detail="Tracking config not found")
if tc.provider_type != provider.type:
raise HTTPException(
packages/server/src/notify_bridge_server/api/providers.py
+5
View File
@@ -97,6 +97,8 @@ async def create_provider(
detail=test_result.get("message", "Cannot connect to Gitea"),
)
# Scheduler: no validation needed (virtual provider)
provider = ServiceProvider(
user_id=user.id,
type=body.type,
@@ -253,6 +255,9 @@ async def test_provider(
gitea = make_gitea_provider(http_session, provider)
return await gitea.test_connection()
if provider.type == "scheduler":
return {"ok": True, "message": "Virtual provider — always available"}
return {"ok": False, "message": f"Unknown provider type: {provider.type}"}
packages/server/src/notify_bridge_server/api/tracking_configs.py
+5 -2
View File
@@ -101,7 +101,10 @@ async def list_configs(
user: User = Depends(get_current_user),
session: AsyncSession = Depends(get_session),
):
query = select(TrackingConfig).where(TrackingConfig.user_id == user.id)
from sqlmodel import or_
query = select(TrackingConfig).where(
or_(TrackingConfig.user_id == user.id, TrackingConfig.user_id == 0)
)
if provider_type:
query = query.where(TrackingConfig.provider_type == provider_type)
result = await session.exec(query)
@@ -167,6 +170,6 @@ def _response(c: TrackingConfig) -> dict:
async def _get(session: AsyncSession, config_id: int, user_id: int) -> TrackingConfig:
config = await session.get(TrackingConfig, config_id)
if not config or config.user_id != user_id:
if not config or (config.user_id != user_id and config.user_id != 0):
raise HTTPException(status_code=404, detail="Tracking config not found")
return config
packages/server/src/notify_bridge_server/api/webhooks.py
+2
View File
@@ -277,6 +277,8 @@ def _event_allowed_by_tracking_config(event: ServiceEvent, tc: TrackingConfig) -
"pr_merged": tc.track_pr_merged,
"pr_commented": tc.track_pr_commented,
"release_published": tc.track_release_published,
# Scheduler events
"scheduled_message": tc.track_scheduled_message,
# Immich events
"assets_added": tc.track_assets_added,
"assets_removed": tc.track_assets_removed,
packages/server/src/notify_bridge_server/database/migrations.py
+1
View File
@@ -150,6 +150,7 @@ async def migrate_schema(engine: AsyncEngine) -> None:
("track_pr_merged", "INTEGER DEFAULT 1"),
("track_pr_commented", "INTEGER DEFAULT 0"),
("track_release_published", "INTEGER DEFAULT 1"),
("track_scheduled_message", "INTEGER DEFAULT 1"),
]
for col_name, col_type in gitea_flags:
if not await _has_column(conn, "tracking_config", col_name):
packages/server/src/notify_bridge_server/database/models.py
+3
View File
@@ -127,6 +127,9 @@ class TrackingConfig(SQLModel, table=True):
track_pr_commented: bool = Field(default=False)
track_release_published: bool = Field(default=True)
# Scheduler event tracking
track_scheduled_message: bool = Field(default=True)
# Immich asset display
track_images: bool = Field(default=True)
track_videos: bool = Field(default=True)
packages/server/src/notify_bridge_server/main.py
+125
View File
@@ -52,6 +52,7 @@ async def lifespan(app: FastAPI):
await migrate_command_slot_locale(engine)
await _seed_default_templates()
await _seed_default_command_templates()
await _seed_default_tracking_configs()
# Configure webhook secret from DB setting (falls back to env var)
from sqlmodel.ext.asyncio.session import AsyncSession as _AS
from .api.app_settings import get_setting as _get_setting
@@ -275,6 +276,85 @@ async def _seed_default_templates():
template=template_text,
))
# --- Seed Scheduler default templates ---
sched_result = await session.exec(
select(TemplateConfig).where(
TemplateConfig.user_id == 0,
TemplateConfig.provider_type == "scheduler",
)
)
sched_configs = sched_result.all()
sched_existing_locales = {
(c.locale if c.locale else "en"): c for c in sched_configs
}
for locale in ("en", "ru"):
sched_slots = load_default_templates(locale, provider_type="scheduler")
if not sched_slots:
continue
if locale not in sched_existing_locales:
from datetime import datetime as _dt2, timezone as _tz2
now2 = _dt2.now(_tz2.utc).isoformat()
name2 = f"Default Scheduler ({locale.upper()})"
desc2 = f"Default Scheduler templates ({locale.upper()})"
col_info2 = (await session.execute(
text("PRAGMA table_info(template_config)")
)).fetchall()
col_names2 = [c[1] for c in col_info2 if c[1] != "id"]
values2 = {}
for col in col_names2:
if col == "user_id":
values2[col] = 0
elif col == "provider_type":
values2[col] = "scheduler"
elif col == "name":
values2[col] = name2
elif col == "description":
values2[col] = desc2
elif col == "created_at":
values2[col] = now2
elif col == "date_format":
values2[col] = "%d.%m.%Y, %H:%M UTC"
elif col == "date_only_format":
values2[col] = "%d.%m.%Y"
elif col == "locale":
values2[col] = locale
else:
values2[col] = ""
cols_str2 = ", ".join(values2.keys())
placeholders2 = ", ".join(f":{k}" for k in values2.keys())
await session.execute(
text(f"INSERT INTO template_config ({cols_str2}) VALUES ({placeholders2})"),
values2,
)
row2 = (await session.execute(text("SELECT last_insert_rowid()"))).scalar()
for slot_name, template_text in sched_slots.items():
session.add(TemplateSlot(
config_id=row2,
slot_name=slot_name,
template=template_text,
))
else:
config = sched_existing_locales[locale]
for slot_name, template_text in sched_slots.items():
slot_result = await session.exec(
select(TemplateSlot).where(
TemplateSlot.config_id == config.id,
TemplateSlot.slot_name == slot_name,
)
)
existing = slot_result.first()
if existing:
existing.template = template_text
session.add(existing)
else:
session.add(TemplateSlot(
config_id=config.id,
slot_name=slot_name,
template=template_text,
))
await session.commit()
@@ -339,6 +419,51 @@ async def _seed_default_command_templates():
await session.commit()
async def _seed_default_tracking_configs():
"""Seed system-owned default tracking configs for each provider type."""
from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession
from .database.engine import get_engine
from .database.models import TrackingConfig
engine = get_engine()
async with AsyncSession(engine) as session:
# Find existing system-owned tracking configs
result = await session.exec(
select(TrackingConfig).where(TrackingConfig.user_id == 0)
)
existing = {c.provider_type: c for c in result.all()}
defaults = [
{
"provider_type": "gitea",
"name": "Default Gitea",
"track_push": True,
"track_issue_opened": True,
"track_issue_closed": True,
"track_issue_commented": False,
"track_pr_opened": True,
"track_pr_closed": True,
"track_pr_merged": True,
"track_pr_commented": False,
"track_release_published": True,
},
{
"provider_type": "scheduler",
"name": "Default Scheduler",
"track_scheduled_message": True,
},
]
for cfg in defaults:
ptype = cfg["provider_type"]
if ptype in existing:
continue
session.add(TrackingConfig(user_id=0, **cfg))
await session.commit()
def run():
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8420)
packages/server/src/notify_bridge_server/services/sample_context.py
+46
View File
@@ -91,4 +91,50 @@ _SAMPLE_CONTEXT = {
"photo_count": 30,
"video_count": 5,
"owner": "Alice",
# Gitea variables (for gitea provider templates)
"sender": "alexei",
"sender_name": "Alexei",
"sender_avatar": "",
"repo_name": "my-project",
"repo_full_name": "alexei/my-project",
"repo_url": "https://gitea.example.com/alexei/my-project",
"repo_description": "Example project",
"branch": "main",
"commits": [{"id": "abc1234567", "short_id": "abc1234", "message": "fix: example commit", "url": "", "author": "Alexei"}],
"commit_count": 1,
"compare_url": "https://gitea.example.com/alexei/my-project/compare/abc...def",
"issue_number": 42,
"issue_title": "Example issue",
"issue_url": "https://gitea.example.com/alexei/my-project/issues/42",
"issue_state": "open",
"issue_body": "",
"issue_labels": ["bug"],
"pr_number": 17,
"pr_title": "Add feature",
"pr_url": "https://gitea.example.com/alexei/my-project/pulls/17",
"pr_state": "open",
"pr_body": "",
"pr_merged": False,
"pr_base": "main",
"pr_head": "feature/example",
"pr_labels": [],
"comment_body": "Looks good!",
"comment_url": "",
"comment_author": "alexei",
"release_tag": "v1.0.0",
"release_name": "Version 1.0.0",
"release_url": "https://gitea.example.com/alexei/my-project/releases/tag/v1.0.0",
"release_body": "Initial release",
"release_draft": False,
"release_prerelease": False,
# Scheduler variables (for scheduler provider templates)
"schedule_name": "Daily Reminder",
"fire_count": 42,
"current_date": "22.03.2026",
"current_time": "09:00",
"current_datetime": "22.03.2026, 09:00 UTC",
"weekday": "Monday",
"custom_vars": {"team": "Engineering", "message": "Time for standup!"},
"team": "Engineering",
"message": "Time for standup!",
}
packages/server/src/notify_bridge_server/services/scheduler.py
+70 -14
View File
@@ -40,7 +40,7 @@ async def _load_tracker_jobs() -> None:
from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession
from ..database.engine import get_engine
from ..database.models import NotificationTracker
from ..database.models import NotificationTracker, ServiceProvider as ServiceProviderModel
engine = get_engine()
scheduler = get_scheduler()
@@ -49,11 +49,35 @@ async def _load_tracker_jobs() -> None:
result = await session.exec(select(NotificationTracker).where(NotificationTracker.enabled == True))
trackers = result.all()
# Pre-load provider types for scheduler detection
provider_types: dict[int, str] = {}
for tracker in trackers:
if tracker.provider_id not in provider_types:
provider = await session.get(ServiceProviderModel, tracker.provider_id)
if provider:
provider_types[tracker.provider_id] = provider.type
for tracker in trackers:
job_id = f"tracker_{tracker.id}"
if scheduler.get_job(job_id):
continue
ptype = provider_types.get(tracker.provider_id, "")
filters = tracker.filters or {}
# Scheduler providers can use cron triggers
if ptype == "scheduler" and filters.get("schedule_type") == "cron":
cron_expr = filters.get("cron_expression", "")
if cron_expr:
try:
_add_cron_job(scheduler, job_id, tracker.id, cron_expr, tracker.name)
continue
except Exception as e:
_LOGGER.error(
"Invalid cron expression for tracker %d (%s): %s — falling back to interval",
tracker.id, tracker.name, e,
)
scheduler.add_job(
_poll_tracker,
"interval",
@@ -65,23 +89,55 @@ async def _load_tracker_jobs() -> None:
_LOGGER.info("Scheduled tracker %d (%s) every %ds", tracker.id, tracker.name, tracker.scan_interval)
async def schedule_tracker(tracker_id: int, interval: int) -> None:
def _add_cron_job(
scheduler: AsyncIOScheduler,
job_id: str,
tracker_id: int,
cron_expression: str,
tracker_name: str,
) -> None:
"""Add a cron-triggered job for a scheduler-type tracker."""
from apscheduler.triggers.cron import CronTrigger
trigger = CronTrigger.from_crontab(cron_expression)
scheduler.add_job(
_poll_tracker,
trigger,
id=job_id,
args=[tracker_id],
replace_existing=True,
)
_LOGGER.info("Scheduled tracker %d (%s) with cron: %s", tracker_id, tracker_name, cron_expression)
async def schedule_tracker(
tracker_id: int,
interval: int,
cron_expression: str | None = None,
) -> None:
"""Add or update a scheduler job for a tracker."""
scheduler = get_scheduler()
job_id = f"tracker_{tracker_id}"
# Remove existing job first to allow trigger type changes
if scheduler.get_job(job_id):
scheduler.reschedule_job(job_id, trigger="interval", seconds=interval)
_LOGGER.info("Rescheduled tracker %d every %ds", tracker_id, interval)
else:
scheduler.add_job(
_poll_tracker,
"interval",
seconds=interval,
id=job_id,
args=[tracker_id],
replace_existing=True,
)
_LOGGER.info("Scheduled tracker %d every %ds", tracker_id, interval)
scheduler.remove_job(job_id)
if cron_expression:
try:
_add_cron_job(scheduler, job_id, tracker_id, cron_expression, f"tracker-{tracker_id}")
return
except Exception as e:
_LOGGER.error("Invalid cron for tracker %d: %s — using interval", tracker_id, e)
scheduler.add_job(
_poll_tracker,
"interval",
seconds=interval,
id=job_id,
args=[tracker_id],
replace_existing=True,
)
_LOGGER.info("Scheduled tracker %d every %ds", tracker_id, interval)
async def unschedule_tracker(tracker_id: int) -> None:
packages/server/src/notify_bridge_server/services/watcher.py
+13
View File
@@ -94,6 +94,8 @@ def _event_allowed_by_config(event: ServiceEvent, tc: TrackingConfig) -> bool:
"pr_merged": tc.track_pr_merged,
"pr_commented": tc.track_pr_commented,
"release_published": tc.track_release_published,
# Scheduler events
"scheduled_message": tc.track_scheduled_message,
}
return flag_map.get(event_type, True)
@@ -210,6 +212,8 @@ async def check_tracker(tracker_id: int) -> dict[str, Any]:
provider_type = provider.type
provider_config = dict(provider.config)
provider_name = provider.name
tracker_name = tracker.name
tracker_filters = dict(tracker.filters) if tracker.filters else {}
collection_ids = list(tracker.collection_ids or [])
# Now create aiohttp session and poll
@@ -235,6 +239,15 @@ async def check_tracker(tracker_id: int) -> dict[str, Any]:
# Gitea is webhook-based — events arrive via /api/webhooks/gitea endpoint.
# The scheduler still calls check_tracker but there's nothing to poll.
return {"status": "ok", "events_detected": 0, "collections_checked": 0}
elif provider_type == "scheduler":
from notify_bridge_core.providers.scheduler import SchedulerServiceProvider
custom_vars = tracker_filters.get("custom_variables", {})
sched = SchedulerServiceProvider(
name=provider_name,
tracker_name=tracker_name,
custom_variables=custom_vars,
)
events, new_state = await sched.poll(collection_ids, state_dict)
else:
return {"status": "error", "reason": f"unsupported provider type: {provider_type}"}