240 lines
13 KiB
Python
240 lines
13 KiB
Python
"""Transactional push outbox, recipient preferences and bounded background delivery.
|
|
|
|
No message bodies, display names, FCM tokens or credentials are stored in the outbox.
|
|
"""
|
|
from datetime import timedelta
|
|
import hashlib
|
|
import logging
|
|
import os
|
|
import threading
|
|
from urllib.parse import quote
|
|
from uuid import uuid4
|
|
|
|
from fastapi import Form, Request
|
|
from fastapi.responses import RedirectResponse
|
|
from i18n import ENGLISH, current_language
|
|
|
|
logger = logging.getLogger(__name__)
|
|
KINDS = ('friend_request', 'direct_message', 'event_invitation')
|
|
TEXT = {
|
|
'friend_request': ('Neue Freundschaftsanfrage', 'Du hast eine neue Freundschaftsanfrage.'),
|
|
'direct_message': ('Neue Nachricht', 'Du hast eine neue Nachricht.'),
|
|
'event_invitation': ('Neue Veranstaltungseinladung', 'Du wurdest zu einer Veranstaltung eingeladen.'),
|
|
}
|
|
|
|
|
|
def enabled():
|
|
return os.environ.get('PUSH_ENABLED', 'false').lower() in ('1', 'true', 'yes')
|
|
|
|
|
|
def save_language(db, user_id, language):
|
|
db.execute('''INSERT INTO notification_preferences(user_id,language) VALUES (%s,%s)
|
|
ON CONFLICT(user_id) DO UPDATE SET language=EXCLUDED.language''',
|
|
(user_id, language if language in ('de', 'en') else 'de'))
|
|
|
|
|
|
def preferences(db, user_id):
|
|
row = db.execute('SELECT language,friend_request,direct_message,event_invitation '
|
|
'FROM notification_preferences WHERE user_id=%s', (user_id,)).fetchone()
|
|
return dict(zip(('language', *KINDS), row or ('de', True, True, True)))
|
|
|
|
|
|
def enqueue(cursor, kind, actor_id, recipient_id, object_id, event_key=None):
|
|
"""Called inside the domain write transaction; disabled means no backlog accumulation."""
|
|
if not enabled() or actor_id == recipient_id:
|
|
return
|
|
if kind not in KINDS:
|
|
raise ValueError('Unknown notification kind')
|
|
cursor.execute('SELECT friend_request,direct_message,event_invitation FROM notification_preferences WHERE user_id=%s',
|
|
(recipient_id,))
|
|
pref = cursor.fetchone()
|
|
if pref and not pref[KINDS.index(kind)]:
|
|
return
|
|
if kind == 'event_invitation':
|
|
# A removed and subsequently re-added invitation replaces its previous pending delivery.
|
|
cursor.execute("UPDATE push_notifications SET state='dropped' WHERE kind='event_invitation' "
|
|
"AND object_id=%s AND recipient_id=%s AND state='pending'", (object_id, recipient_id))
|
|
cursor.execute('''SELECT p.id,p.session_id,p.token FROM push_devices p
|
|
JOIN sessions s ON s.id=p.session_id
|
|
WHERE p.user_id=%s AND s.user_id=%s AND s.expires_at>CURRENT_TIMESTAMP
|
|
AND p.last_seen_at>CURRENT_TIMESTAMP-INTERVAL '90 days' ''', (recipient_id, recipient_id))
|
|
for device_id, session_id, token in cursor.fetchall():
|
|
cursor.execute('''INSERT INTO push_notifications
|
|
(id,device_id,session_id,recipient_id,actor_id,kind,object_id,event_key,token_hash)
|
|
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s) ON CONFLICT(device_id,event_key) DO NOTHING''',
|
|
(uuid4(), device_id, session_id, recipient_id, actor_id, kind, object_id,
|
|
event_key or f'{kind}:{object_id}', hashlib.sha256(token.encode()).hexdigest()))
|
|
|
|
|
|
def destination(db, kind, actor_id, recipient_id, object_id):
|
|
"""Recheck current authorization and event state immediately before sending/opening."""
|
|
if db.execute('''SELECT 1 FROM user_blocks WHERE
|
|
(blocker_id=%s AND blocked_id=%s) OR (blocker_id=%s AND blocked_id=%s)''',
|
|
(actor_id, recipient_id, recipient_id, actor_id)).fetchone():
|
|
return None
|
|
actor = db.execute('SELECT username,is_admin FROM users WHERE id=%s', (actor_id,)).fetchone()
|
|
recipient = db.execute('SELECT is_admin FROM users WHERE id=%s', (recipient_id,)).fetchone()
|
|
if not actor or not recipient:
|
|
return None
|
|
if kind == 'friend_request':
|
|
valid = db.execute("SELECT 1 FROM friendships WHERE id=%s AND requester_id=%s AND addressee_id=%s AND status='pending'",
|
|
(object_id, actor_id, recipient_id)).fetchone()
|
|
return '/users/' + quote(actor[0], safe='') if valid else None
|
|
if kind == 'direct_message':
|
|
valid = db.execute('SELECT 1 FROM direct_messages WHERE id=%s AND sender_id=%s AND recipient_id=%s AND read_at IS NULL',
|
|
(object_id, actor_id, recipient_id)).fetchone()
|
|
allowed = actor[1] or recipient[0] or db.execute("""SELECT 1 FROM friendships WHERE status='accepted'
|
|
AND ((requester_id=%s AND addressee_id=%s) OR (requester_id=%s AND addressee_id=%s))""",
|
|
(actor_id, recipient_id, recipient_id, actor_id)).fetchone()
|
|
return '/messages/' + quote(actor[0], safe='') + '#latest' if valid and allowed else None
|
|
if kind == 'event_invitation':
|
|
valid = db.execute('''SELECT 1 FROM event_invitations i JOIN concerts c ON c.id=i.concert_id
|
|
WHERE i.concert_id=%s AND i.user_id=%s AND i.invited_by=%s AND i.viewed_at IS NULL
|
|
AND (c.visibility IN ('public','private') OR c.created_by=%s OR %s OR EXISTS (
|
|
SELECT 1 FROM friendships f WHERE f.status='accepted' AND
|
|
((f.requester_id=c.created_by AND f.addressee_id=%s) OR
|
|
(f.addressee_id=c.created_by AND f.requester_id=%s))))''',
|
|
(object_id, recipient_id, actor_id, recipient_id, recipient[0], recipient_id, recipient_id)).fetchone()
|
|
return '/concerts/' + str(object_id) if valid else None
|
|
return None
|
|
|
|
|
|
class DeliveryError(Exception):
|
|
def __init__(self, code):
|
|
self.code = code
|
|
super().__init__(code)
|
|
|
|
|
|
class FirebaseSender:
|
|
def __init__(self):
|
|
self.app = None
|
|
|
|
def send(self, token, title, body, data, tag):
|
|
# Lazy import/init: missing credentials never prevent the core app from starting.
|
|
try:
|
|
import firebase_admin
|
|
from firebase_admin import credentials, messaging
|
|
if self.app is None:
|
|
path = os.environ.get('GOOGLE_APPLICATION_CREDENTIALS', '')
|
|
project = os.environ.get('FIREBASE_PROJECT_ID', '')
|
|
if not path or not project:
|
|
raise DeliveryError('configuration')
|
|
credential = credentials.Certificate(path)
|
|
if credential.project_id != project:
|
|
raise DeliveryError('configuration')
|
|
self.app = firebase_admin.initialize_app(credential, {'projectId': project, 'httpTimeout': 10},
|
|
name='metalcircle-push-' + uuid4().hex)
|
|
messaging.send(messaging.Message(token=token, notification=messaging.Notification(title=title, body=body),
|
|
data=data, android=messaging.AndroidConfig(priority='high', ttl=timedelta(minutes=5),
|
|
notification=messaging.AndroidNotification(tag=tag, icon='ic_notification',
|
|
sound='default', visibility='private'))), app=self.app)
|
|
except DeliveryError:
|
|
raise
|
|
except Exception as error:
|
|
# Never log SDK errors: they may contain requests, credentials or FCM tokens.
|
|
code = getattr(error, 'code', '')
|
|
name = type(error).__name__
|
|
if name == 'UnregisteredError':
|
|
raise DeliveryError('unregistered') from None
|
|
if code in ('UNAVAILABLE', 'INTERNAL', 'DEADLINE_EXCEEDED', 'RESOURCE_EXHAUSTED'):
|
|
raise DeliveryError('transient') from None
|
|
if isinstance(error, (OSError, ValueError, ImportError)) or code in ('UNAUTHENTICATED', 'PERMISSION_DENIED'):
|
|
raise DeliveryError('configuration') from None
|
|
raise DeliveryError('permanent') from None
|
|
|
|
|
|
class PushWorker:
|
|
def __init__(self, get_db, sender=None):
|
|
self.get_db = get_db
|
|
self.sender = sender or FirebaseSender()
|
|
self.stop_event = threading.Event()
|
|
self.thread = None
|
|
|
|
def start(self):
|
|
if enabled():
|
|
self.thread = threading.Thread(target=self.run, name='metalcircle-push', daemon=True)
|
|
self.thread.start()
|
|
|
|
def stop(self):
|
|
self.stop_event.set()
|
|
if self.thread:
|
|
self.thread.join(timeout=15)
|
|
|
|
def run(self):
|
|
while not self.stop_event.is_set():
|
|
try:
|
|
busy = self.deliver_one()
|
|
except Exception:
|
|
logger.warning('Push worker: database_or_processing_failure')
|
|
busy = False
|
|
self.stop_event.wait(0.1 if busy else 3)
|
|
|
|
def deliver_one(self):
|
|
with self.get_db() as db:
|
|
db.execute("DELETE FROM push_notifications WHERE created_at<CURRENT_TIMESTAMP-INTERVAL '7 days'")
|
|
row = db.execute('''SELECT id,device_id,session_id,recipient_id,actor_id,kind,object_id,token_hash,attempts,
|
|
expires_at>CURRENT_TIMESTAMP FROM push_notifications WHERE state='pending'
|
|
AND available_at<=CURRENT_TIMESTAMP ORDER BY available_at LIMIT 1 FOR UPDATE SKIP LOCKED''').fetchone()
|
|
if not row:
|
|
return False
|
|
job, device_id, session_id, recipient, actor, kind, obj, token_hash, attempts, fresh = row
|
|
# Match registration's lock order. NOWAIT avoids a deadlock with session deletion's cascade.
|
|
session = db.execute('SELECT token_hash FROM sessions WHERE id=%s AND user_id=%s '
|
|
'AND expires_at>CURRENT_TIMESTAMP FOR UPDATE NOWAIT', (session_id, recipient)).fetchone()
|
|
device = db.execute('SELECT token FROM push_devices WHERE id=%s AND session_id=%s AND user_id=%s '
|
|
'FOR UPDATE NOWAIT', (device_id, session_id, recipient)).fetchone()
|
|
prefs = preferences(db, recipient)
|
|
target = destination(db, kind, actor, recipient, obj) if fresh and prefs[kind] else None
|
|
if not session or not device or not target or hashlib.sha256(device[0].encode()).hexdigest() != token_hash:
|
|
db.execute("UPDATE push_notifications SET state='dropped' WHERE id=%s", (job,))
|
|
return True
|
|
title, body = TEXT[kind]
|
|
if prefs['language'] == 'en':
|
|
title, body = ENGLISH.get(title, title), ENGLISH.get(body, body)
|
|
try:
|
|
self.sender.send(device[0], title, body,
|
|
{'notification_id': str(job), 'session_tag': session[0]}, str(job))
|
|
except DeliveryError as error:
|
|
logger.warning('Push delivery failed: %s', error.code)
|
|
if error.code == 'unregistered':
|
|
db.execute('DELETE FROM push_devices WHERE id=%s AND token=%s', (device_id, device[0]))
|
|
elif error.code in ('transient', 'configuration') and attempts < 3:
|
|
db.execute("UPDATE push_notifications SET attempts=attempts+1, available_at=CURRENT_TIMESTAMP + %s * INTERVAL '1 second' WHERE id=%s",
|
|
(60 * 2 ** attempts, job))
|
|
else:
|
|
db.execute("UPDATE push_notifications SET state='failed',attempts=attempts+1 WHERE id=%s", (job,))
|
|
else:
|
|
db.execute("UPDATE push_notifications SET state='sent',attempts=attempts+1 WHERE id=%s", (job,))
|
|
return True
|
|
|
|
|
|
def register_routes(app, get_db, get_user, cookie_name):
|
|
@app.post('/profile/notifications')
|
|
def update_preferences(request: Request, friend_request: bool = Form(False),
|
|
direct_message: bool = Form(False), event_invitation: bool = Form(False)):
|
|
user = get_user(request)
|
|
with get_db() as db:
|
|
save_language(db, user['id'], current_language.get())
|
|
db.execute('''UPDATE notification_preferences SET friend_request=%s,direct_message=%s,event_invitation=%s
|
|
WHERE user_id=%s''', (friend_request, direct_message, event_invitation, user['id']))
|
|
# Turning a category off cancels pending notifications, even if quickly enabled again.
|
|
disabled = [kind for kind, value in zip(KINDS, (friend_request, direct_message, event_invitation)) if not value]
|
|
db.execute("UPDATE push_notifications SET state='dropped' WHERE recipient_id=%s AND state='pending' AND kind=ANY(%s)",
|
|
(user['id'], disabled))
|
|
return RedirectResponse('/profile?saved=notifications#notification-settings', status_code=303)
|
|
|
|
@app.get('/notifications/{identifier}')
|
|
def open_notification(request: Request, identifier: str):
|
|
from uuid import UUID
|
|
try:
|
|
identifier = UUID(identifier)
|
|
except ValueError:
|
|
return RedirectResponse('/', status_code=303)
|
|
user = get_user(request)
|
|
with get_db() as db:
|
|
row = db.execute('''SELECT kind,actor_id,object_id FROM push_notifications n JOIN sessions s ON s.id=n.session_id
|
|
WHERE n.id=%s AND n.recipient_id=%s AND s.token_hash=%s AND s.expires_at>CURRENT_TIMESTAMP''',
|
|
(identifier, user['id'], hashlib.sha256(request.cookies.get(cookie_name, '').encode()).hexdigest())).fetchone()
|
|
target = destination(db, row[0], row[1], user['id'], row[2]) if row else None
|
|
return RedirectResponse(target or '/', status_code=303)
|