Feat: Allow for notification support.

app.py

@@ -30,6 +30,7 @@
 from routes.digicard import digicard_bp, set_oidc as set_oidc_digicard
 from routes.misc import misc_routes_bp
 from routes.coupons_admin import coupons_admin_bp
+from routes.notifications_admin import notifications_admin_bp
 from utils.cache import init_cache, generate_top_page_cache
 from utils.achievements import sync_achievements
 from utils.db import init_db
@@ -110,6 +111,7 @@ def utility_processor():
 app.register_blueprint(digicard_bp)
 app.register_blueprint(misc_routes_bp)
 app.register_blueprint(coupons_admin_bp)
+app.register_blueprint(notifications_admin_bp)
 
 # Background scheduler for cache generation
 scheduler = BackgroundScheduler()

config_example.py

@@ -31,3 +31,12 @@
     }
 }
 oidc_logout_url = ""
+
+# Web Push notifications.
+vapid_subject = "mailto:change-me@example.com"
+vapid_public_key = ""
+vapid_private_key = ""
+
+notifications_shared_secret = ""
+
+accountmanager_url = "http://localhost:9011"

migrations/env.py

@@ -2,11 +2,11 @@
 from sqlalchemy import create_engine, pool
 
 import config as app_config
-from utils.db import db
+from utils.db import CHECKOUT_BIND, db
 
 config = context.config
 
-target_metadata = db.metadata
+target_metadata = db.metadatas[CHECKOUT_BIND]
 
 url = app_config.checkout_db_url
 

migrations/versions/3404e6b4acfe_add_notification_preferences.py

@@ -0,0 +1,34 @@
+"""add notification_preferences
+
+Revision ID: 3404e6b4acfe
+Revises: b8f326057d6d
+Create Date: 2026-09-01 21:59:25.577785
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = '3404e6b4acfe'
+down_revision = 'b8f326057d6d'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('notification_preferences',
+    sa.Column('username', sa.Text(), nullable=False),
+    sa.Column('category', sa.Text(), nullable=False),
+    sa.Column('enabled', sa.Boolean(), server_default=sa.text('true'), nullable=False),
+    sa.PrimaryKeyConstraint('username', 'category')
+    )
+    op.drop_index(op.f('idx_push_subscriptions_username'), table_name='push_subscriptions')
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_index(op.f('idx_push_subscriptions_username'), 'push_subscriptions', ['username'], unique=False)
+    op.drop_table('notification_preferences')
+    # ### end Alembic commands ###

migrations/versions/b8f326057d6d_add_push_subscriptions_table.py

@@ -0,0 +1,50 @@
+"""add push_subscriptions table
+
+Revision ID: b8f326057d6d
+Revises: 0001
+Create Date: 2026-09-01 20:26:26.866492
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+revision = "b8f326057d6d"
+down_revision = "0001"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "push_subscriptions",
+        sa.Column("endpoint", sa.Text(), nullable=False),
+        sa.Column("username", sa.Text(), nullable=False),
+        sa.Column("p256dh", sa.Text(), nullable=False),
+        sa.Column("auth", sa.Text(), nullable=False),
+        sa.Column(
+            "created_at",
+            postgresql.TIMESTAMP(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.PrimaryKeyConstraint("endpoint"),
+    )
+    op.create_index(
+        op.f("ix_push_subscriptions_username"),
+        "push_subscriptions",
+        ["username"],
+        unique=False,
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index(
+        op.f("ix_push_subscriptions_username"), table_name="push_subscriptions"
+    )
+    op.drop_table("push_subscriptions")
+    # ### end Alembic commands ###

requirements.txt

@@ -11,3 +11,5 @@ colorgram.py
 redis
 sqlalchemy
 alembic
+requests
+pywebpush

routes/misc.py

@@ -1,12 +1,21 @@
-from flask import Blueprint, render_template, request, redirect, url_for
+import hmac
+from flask import Blueprint, jsonify, render_template, request, redirect, url_for
+from urllib.parse import urlparse
 import random
+from sqlalchemy import func
+from utils.db import PushSubscription, db
+from utils.push import CATEGORIES, send_push_to_all, send_push_to_user
+
+
 import config
 from utils.utils import (
     search_authentik_users_by_name,
     fetch_authentik_users,
     _run_query,
 )
 
+MAX_SUBSCRIPTIONS_PER_USER = 10
+
 misc_routes_bp = Blueprint("misc_routes", __name__, url_prefix="")
 
 
@@ -183,6 +192,202 @@ def search():
     )
 
 
+def _shared_secret_authorized():
+    auth = request.headers.get("Authorization", "")
+    expected = f"Bearer {config.notifications_shared_secret}"
+    return bool(config.notifications_shared_secret) and hmac.compare_digest(
+        auth, expected
+    )
+
+
+@misc_routes_bp.route(
+    "/notifications/test", methods=["POST"], endpoint="test_notification"
+)
+def test_notification():
+    if not _shared_secret_authorized():
+        return jsonify({"error": "unauthorized"}), 401
+
+    data = request.get_json(silent=True) or {}
+    username = (data.get("username") or "").strip()
+    if not username:
+        return jsonify({"error": "username is required"}), 400
+
+    sent = send_push_to_user(
+        username,
+        title="Checkout Test Notification",
+        body="Push notifications are working!",
+        url="/",
+    )
+    return jsonify({"sent": sent})
+
+
+@misc_routes_bp.route(
+    "/api/notifications/send", methods=["POST"], endpoint="send_notification"
+)
+def send_notification():
+    """
+    Body: {"username": "...", "category": "...", "title": "...", "body": "...",
+           "url": "...", "tag": "...", "icon": "..."}"""
+    if not _shared_secret_authorized():
+        return jsonify({"error": "unauthorized"}), 401
+
+    data = request.get_json(silent=True) or {}
+    title = (data.get("title") or "").strip()
+    if not title:
+        return jsonify({"error": "title is required"}), 400
+
+    category = (data.get("category") or "").strip() or None
+    if category and category not in CATEGORIES:
+        return jsonify({"error": "unknown category"}), 400
+
+    username = (data.get("username") or "").strip()
+    if username:
+        sent = send_push_to_user(
+            username,
+            title=title,
+            body=data.get("body", ""),
+            url=data.get("url", "/"),
+            tag=data.get("tag"),
+            category=category,
+            icon=data.get("icon"),
+        )
+    else:
+        sent = send_push_to_all(
+            title=title,
+            body=data.get("body", ""),
+            url=data.get("url", "/"),
+            tag=data.get("tag"),
+            category=category,
+            icon=data.get("icon"),
+        )
+    return jsonify({"sent": sent})
+
+
+def _safe_return_target():
+    target = request.args.get("return", "/")
+    if target.startswith("/") and not target.startswith("//"):
+        return target
+
+    trusted_origin = _accountmanager_origin()
+    if trusted_origin:
+        parsed = urlparse(target)
+        if parsed.scheme in ("http", "https") and (
+            f"{parsed.scheme}://{parsed.netloc}" == trusted_origin
+        ):
+            return target
+    return "/"
+
+
+def _accountmanager_origin():
+    parsed = urlparse(config.accountmanager_url or "")
+    if parsed.scheme and parsed.netloc:
+        return f"{parsed.scheme}://{parsed.netloc}"
+    return None
+
+
+@misc_routes_bp.route("/notifications/enable", endpoint="enable_notifications")
+def enable_notifications():
+    user_info = get_logged_in_user_info()
+    if not user_info:
+        return redirect(url_for("oidc_auth.login"))
+
+    configured = bool(config.vapid_public_key)
+    return render_template(
+        "notifications_enable.html",
+        public_key=config.vapid_public_key,
+        return_url=_safe_return_target(),
+        status_message=(
+            None
+            if configured
+            else "Notifications are not configured on this server yet."
+        ),
+    )
+
+
+@misc_routes_bp.route("/notifications/disable", endpoint="disable_notifications")
+def disable_notifications():
+    user_info = get_logged_in_user_info()
+    if not user_info:
+        return redirect(url_for("oidc_auth.login"))
+    return render_template(
+        "notifications_disable.html", return_url=_safe_return_target()
+    )
+
+
+@misc_routes_bp.route(
+    "/notifications/subscribe", methods=["POST"], endpoint="subscribe_notification"
+)
+def subscribe_notification():
+    user_info = get_logged_in_user_info()
+    username = (user_info or {}).get("username", "")
+    if not username:
+        return jsonify({"error": "Not logged in"}), 401
+
+    data = request.get_json(silent=True) or {}
+    endpoint = (data.get("endpoint") or "").strip()
+    keys = data.get("keys") or {}
+    p256dh = keys.get("p256dh")
+    auth = keys.get("auth")
+    if not endpoint or not p256dh or not auth:
+        return jsonify({"error": "Invalid subscription"}), 400
+
+    existing = db.session.get(PushSubscription, endpoint)
+    if existing:
+        existing.username = username
+        existing.p256dh = p256dh
+        existing.auth = auth
+    else:
+        count = db.session.execute(
+            db.select(func.count())
+            .select_from(PushSubscription)
+            .where(PushSubscription.username == username)
+        ).scalar()
+        if count >= MAX_SUBSCRIPTIONS_PER_USER:
+            oldest = (
+                db.session.execute(
+                    db.select(PushSubscription)
+                    .where(PushSubscription.username == username)
+                    .order_by(PushSubscription.created_at)
+                    .limit(count - MAX_SUBSCRIPTIONS_PER_USER + 1)
+                )
+                .scalars()
+                .all()
+            )
+            for row in oldest:
+                db.session.delete(row)
+        db.session.add(
+            PushSubscription(
+                endpoint=endpoint, username=username, p256dh=p256dh, auth=auth
+            )
+        )
+    db.session.commit()
+    return jsonify({"ok": True})
+
+
+@misc_routes_bp.route(
+    "/notifications/unsubscribe", methods=["POST"], endpoint="unsubscribe_notification"
+)
+def unsubscribe_notification():
+    user_info = get_logged_in_user_info()
+    username = (user_info or {}).get("username", "")
+    if not username:
+        return jsonify({"error": "Not logged in"}), 401
+
+    data = request.get_json(silent=True) or {}
+    endpoint = (data.get("endpoint") or "").strip()
+    if not endpoint:
+        return jsonify({"error": "Invalid subscription"}), 400
+
+    db.session.execute(
+        db.delete(PushSubscription).where(
+            PushSubscription.endpoint == endpoint,
+            PushSubscription.username == username,
+        )
+    )
+    db.session.commit()
+    return jsonify({"ok": True})
+
+
 @misc_routes_bp.route("/logout", endpoint="logout")
 def logout():
     """Logout user and redirect to login page"""