Ryan Kegel
ad2bdf4c4f
All checks were successful
Chore App Build, Test, and Push Docker Images / build-and-push (push) Successful in 2m14s
- Implemented push subscription API with tests for subscribing and unsubscribing users. - Created web push notification tests triggered by child actions. - Added digest scheduler to send email digests to users at 9 PM local time. - Developed utility functions for creating and validating digest action tokens. - Integrated web push sender to handle sending notifications to users. - Added service worker for handling push notifications in the frontend. - Created a push opt-in component for user notification preferences. - Implemented tests for the push opt-in component to ensure correct behavior. - Updated frontend services to manage push subscriptions and permissions.
39 lines
1.5 KiB
Python
39 lines
1.5 KiB
Python
from tinydb import Query
|
|
from db.db import push_subscriptions_db
|
|
from models.push_subscription import PushSubscription
|
|
|
|
|
|
def get_subscriptions_by_user(user_id: str) -> list[PushSubscription]:
|
|
"""Return all push subscriptions for a user."""
|
|
Q = Query()
|
|
results = push_subscriptions_db.search(Q.user_id == user_id)
|
|
return [PushSubscription.from_dict(r) for r in results]
|
|
|
|
|
|
def upsert_subscription(user_id: str, endpoint: str, keys: dict) -> PushSubscription:
|
|
"""Insert or update a subscription for the given user+endpoint pair."""
|
|
Q = Query()
|
|
existing = push_subscriptions_db.get((Q.user_id == user_id) & (Q.endpoint == endpoint))
|
|
if existing:
|
|
sub = PushSubscription.from_dict(existing)
|
|
sub.keys = keys
|
|
sub.touch()
|
|
push_subscriptions_db.update(sub.to_dict(), (Q.user_id == user_id) & (Q.endpoint == endpoint))
|
|
return sub
|
|
sub = PushSubscription(user_id=user_id, endpoint=endpoint, keys=keys)
|
|
push_subscriptions_db.insert(sub.to_dict())
|
|
return sub
|
|
|
|
|
|
def delete_by_endpoint(user_id: str, endpoint: str) -> int:
|
|
"""Remove the subscription with the given endpoint for this user. Returns count removed."""
|
|
Q = Query()
|
|
removed = push_subscriptions_db.remove((Q.user_id == user_id) & (Q.endpoint == endpoint))
|
|
return len(removed)
|
|
|
|
|
|
def delete_subscription_by_id(subscription_id: str) -> None:
|
|
"""Remove a subscription by its ID (used when push delivery fails)."""
|
|
Q = Query()
|
|
push_subscriptions_db.remove(Q.id == subscription_id)
|