623 lines
25 KiB
Python
623 lines
25 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import os
|
|
import sqlite3
|
|
import time
|
|
from collections.abc import Iterable, Mapping
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from src.private_state import connect_private_sqlite
|
|
from src.state_encryption import (
|
|
PrivateStateCipher,
|
|
decode_private_state_encryption_key,
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PushDelivery:
|
|
session_id: str
|
|
subscription: dict
|
|
thread_revisions: tuple[tuple[int, str], ...]
|
|
digest_revisions: tuple[tuple[int, str], ...] = ()
|
|
|
|
@property
|
|
def thread_ids(self) -> tuple[int, ...]:
|
|
return tuple(thread_id for thread_id, _revision in self.thread_revisions)
|
|
|
|
@property
|
|
def digest_ids(self) -> tuple[int, ...]:
|
|
return tuple(thread_id for thread_id, _revision in self.digest_revisions)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DeadlineReminderDevice:
|
|
session_id: str
|
|
subscription: dict
|
|
timezone: str
|
|
reminder_hour: int
|
|
reminder_days: int
|
|
delivered_local_day: str | None
|
|
snoozed_until: float | None
|
|
|
|
|
|
class DisabledPushSubscriptionStore:
|
|
"""No-persistence store used when Web Push is not configured."""
|
|
|
|
def is_subscribed(self, session_id: str) -> bool:
|
|
return False
|
|
|
|
def subscription_for_session(self, session_id: str) -> dict | None:
|
|
return None
|
|
|
|
def deadline_preferences(
|
|
self, session_id: str, *, now: float | None = None
|
|
) -> dict:
|
|
return {
|
|
"enabled": False,
|
|
"timezone": "UTC",
|
|
"reminder_hour": 9,
|
|
"reminder_days": 2,
|
|
"snoozed_until": None,
|
|
}
|
|
|
|
def deadline_reminder_devices(self) -> list[DeadlineReminderDevice]:
|
|
return []
|
|
|
|
def claim_unseen(self, thread_revisions) -> list[PushDelivery]:
|
|
return []
|
|
|
|
def acquire_dispatch_lease(self, *args, **kwargs) -> bool:
|
|
return False
|
|
|
|
def release_dispatch_lease(self, *args, **kwargs) -> bool:
|
|
return False
|
|
|
|
def snooze_deadline_reminder(self, *args, **kwargs) -> bool:
|
|
return False
|
|
|
|
def upsert(self, *args, **kwargs) -> None:
|
|
raise RuntimeError("push notifications are not configured")
|
|
|
|
def delete_session(self, *args, **kwargs) -> None:
|
|
return None
|
|
|
|
def delete_all(self, *args, **kwargs) -> None:
|
|
return None
|
|
|
|
def set_deadline_preferences(self, *args, **kwargs) -> None:
|
|
return None
|
|
|
|
def clear_deadline_snooze(self, *args, **kwargs) -> None:
|
|
return None
|
|
|
|
def mark_deadline_reminder_delivered(self, *args, **kwargs) -> None:
|
|
return None
|
|
|
|
def reconcile_unread(self, *args, **kwargs) -> None:
|
|
return None
|
|
|
|
def mark_digest_pending(self, *args, **kwargs) -> None:
|
|
return None
|
|
|
|
def mark_delivered(self, *args, **kwargs) -> None:
|
|
return None
|
|
|
|
def delivery_health(self, *args, **kwargs) -> dict:
|
|
return {}
|
|
|
|
def mark_delivery_failed(self, *args, **kwargs) -> None:
|
|
return None
|
|
|
|
def mark_delivery_succeeded(self, *args, **kwargs) -> None:
|
|
return None
|
|
|
|
|
|
def build_push_subscription_store(
|
|
path: str | Path, *, push_enabled: bool
|
|
) -> PushSubscriptionStore | DisabledPushSubscriptionStore:
|
|
if not push_enabled:
|
|
return DisabledPushSubscriptionStore()
|
|
return PushSubscriptionStore(path)
|
|
|
|
|
|
def _revisions(
|
|
values: Mapping[int, str] | Iterable[int | tuple[int, str]],
|
|
) -> tuple[tuple[int, str], ...]:
|
|
if isinstance(values, Mapping):
|
|
items = list(values.items())
|
|
else:
|
|
items = []
|
|
for value in values:
|
|
if isinstance(value, tuple):
|
|
items.append(value)
|
|
else:
|
|
items.append((value, ""))
|
|
normalized = {}
|
|
for thread_id, revision in items:
|
|
thread_id = int(thread_id)
|
|
if thread_id > 0:
|
|
normalized[thread_id] = str(revision or "")
|
|
return tuple(sorted(normalized.items()))
|
|
|
|
|
|
class PushSubscriptionStore:
|
|
"""Durable, device-bound Web Push subscriptions and delivery deduplication."""
|
|
|
|
def __init__(self, path: str | Path, *, encryption_key: bytes | None = None):
|
|
self.path = Path(path)
|
|
key = encryption_key
|
|
if key is None:
|
|
key = decode_private_state_encryption_key(
|
|
os.getenv("STACKCHAIN_PUSH_STATE_ENCRYPTION_KEY", "")
|
|
)
|
|
self._encryption_key = key
|
|
self._cipher = PrivateStateCipher(key, store="push-subscriptions")
|
|
with self._connect() as connection:
|
|
connection.executescript(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS push_subscriptions (
|
|
session_id TEXT PRIMARY KEY,
|
|
endpoint TEXT NOT NULL UNIQUE,
|
|
subscription_json TEXT NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS push_deliveries (
|
|
session_id TEXT NOT NULL,
|
|
thread_id INTEGER NOT NULL,
|
|
revision TEXT NOT NULL DEFAULT '',
|
|
PRIMARY KEY (session_id, thread_id),
|
|
FOREIGN KEY (session_id) REFERENCES push_subscriptions(session_id)
|
|
ON DELETE CASCADE
|
|
);
|
|
CREATE TABLE IF NOT EXISTS push_dispatch_lease (
|
|
channel TEXT PRIMARY KEY,
|
|
owner TEXT NOT NULL,
|
|
expires_at REAL NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS push_digest_pending (
|
|
session_id TEXT NOT NULL,
|
|
thread_id INTEGER NOT NULL,
|
|
revision TEXT NOT NULL DEFAULT '',
|
|
PRIMARY KEY (session_id, thread_id),
|
|
FOREIGN KEY (session_id) REFERENCES push_subscriptions(session_id)
|
|
ON DELETE CASCADE
|
|
);
|
|
CREATE TABLE IF NOT EXISTS push_deadline_preferences (
|
|
session_id TEXT PRIMARY KEY,
|
|
enabled INTEGER NOT NULL DEFAULT 0,
|
|
timezone TEXT NOT NULL DEFAULT 'UTC',
|
|
reminder_hour INTEGER NOT NULL DEFAULT 9,
|
|
reminder_days INTEGER NOT NULL DEFAULT 2,
|
|
delivered_local_day TEXT,
|
|
snoozed_until REAL,
|
|
FOREIGN KEY (session_id) REFERENCES push_subscriptions(session_id)
|
|
ON DELETE CASCADE
|
|
);
|
|
CREATE TABLE IF NOT EXISTS push_delivery_health (
|
|
session_id TEXT NOT NULL,
|
|
channel TEXT NOT NULL,
|
|
consecutive_failures INTEGER NOT NULL DEFAULT 0,
|
|
last_attempted_at REAL NOT NULL,
|
|
last_succeeded_at REAL,
|
|
reason TEXT,
|
|
PRIMARY KEY (session_id, channel),
|
|
FOREIGN KEY (session_id) REFERENCES push_subscriptions(session_id)
|
|
ON DELETE CASCADE
|
|
);
|
|
"""
|
|
)
|
|
delivery_columns = {
|
|
row[1] for row in connection.execute("PRAGMA table_info(push_deliveries)")
|
|
}
|
|
if "revision" not in delivery_columns:
|
|
connection.execute(
|
|
"ALTER TABLE push_deliveries ADD COLUMN revision TEXT NOT NULL DEFAULT '*'"
|
|
)
|
|
digest_columns = {
|
|
row[1]
|
|
for row in connection.execute("PRAGMA table_info(push_digest_pending)")
|
|
}
|
|
if "revision" not in digest_columns:
|
|
connection.execute(
|
|
"ALTER TABLE push_digest_pending ADD COLUMN revision TEXT NOT NULL DEFAULT '*'"
|
|
)
|
|
lease_columns = {
|
|
row[1] for row in connection.execute("PRAGMA table_info(push_dispatch_lease)")
|
|
}
|
|
if "singleton" in lease_columns:
|
|
connection.executescript(
|
|
"""
|
|
ALTER TABLE push_dispatch_lease RENAME TO push_dispatch_lease_legacy;
|
|
CREATE TABLE push_dispatch_lease (
|
|
channel TEXT PRIMARY KEY,
|
|
owner TEXT NOT NULL,
|
|
expires_at REAL NOT NULL
|
|
);
|
|
INSERT INTO push_dispatch_lease(channel, owner, expires_at)
|
|
SELECT 'unread', owner, expires_at
|
|
FROM push_dispatch_lease_legacy WHERE singleton = 1;
|
|
DROP TABLE push_dispatch_lease_legacy;
|
|
"""
|
|
)
|
|
preference_columns = {
|
|
row[1]
|
|
for row in connection.execute("PRAGMA table_info(push_deadline_preferences)")
|
|
}
|
|
if "reminder_days" not in preference_columns:
|
|
connection.execute(
|
|
"ALTER TABLE push_deadline_preferences ADD COLUMN reminder_days INTEGER NOT NULL DEFAULT 2"
|
|
)
|
|
if "snoozed_until" not in preference_columns:
|
|
connection.execute(
|
|
"ALTER TABLE push_deadline_preferences ADD COLUMN snoozed_until REAL"
|
|
)
|
|
rows = connection.execute(
|
|
"SELECT session_id, endpoint, subscription_json FROM push_subscriptions"
|
|
).fetchall()
|
|
for session_id, endpoint, payload in rows:
|
|
subscription, plaintext = self._cipher.open(payload, binding=session_id)
|
|
if not isinstance(subscription, dict) or not subscription.get("endpoint"):
|
|
raise ValueError("push subscription payload is invalid")
|
|
endpoint_index = self._endpoint_index(subscription["endpoint"])
|
|
if plaintext or endpoint != endpoint_index:
|
|
connection.execute(
|
|
"UPDATE push_subscriptions SET endpoint = ?, subscription_json = ? WHERE session_id = ?",
|
|
(
|
|
endpoint_index,
|
|
self._cipher.seal(subscription, binding=session_id),
|
|
session_id,
|
|
),
|
|
)
|
|
|
|
def _endpoint_index(self, endpoint: str) -> str:
|
|
return hmac.new(
|
|
self._encryption_key,
|
|
b"stackchain:push-endpoint:v1\0" + endpoint.encode(),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
|
|
def _connect(self):
|
|
connection = connect_private_sqlite(self.path, timeout=2)
|
|
connection.execute("PRAGMA foreign_keys = ON")
|
|
return connection
|
|
|
|
def acquire_dispatch_lease(
|
|
self, owner: str, *, channel: str = "unread", now: float, lease_seconds: float
|
|
) -> bool:
|
|
with self._connect() as connection:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
current = connection.execute(
|
|
"SELECT owner, expires_at FROM push_dispatch_lease WHERE channel = ?",
|
|
(channel,),
|
|
).fetchone()
|
|
if current is not None and current[0] != owner and current[1] > now:
|
|
return False
|
|
connection.execute(
|
|
"""INSERT INTO push_dispatch_lease(channel, owner, expires_at)
|
|
VALUES (?, ?, ?)
|
|
ON CONFLICT(channel) DO UPDATE SET
|
|
owner = excluded.owner, expires_at = excluded.expires_at""",
|
|
(channel, owner, now + lease_seconds),
|
|
)
|
|
return True
|
|
|
|
def release_dispatch_lease(self, owner: str, *, channel: str = "unread") -> bool:
|
|
with self._connect() as connection:
|
|
result = connection.execute(
|
|
"DELETE FROM push_dispatch_lease WHERE channel = ? AND owner = ?",
|
|
(channel, owner),
|
|
)
|
|
return result.rowcount == 1
|
|
|
|
def upsert(self, session_id: str, subscription: dict) -> None:
|
|
endpoint = subscription["endpoint"]
|
|
endpoint_index = self._endpoint_index(endpoint)
|
|
encoded = self._cipher.seal(subscription, binding=session_id)
|
|
with self._connect() as connection:
|
|
connection.execute(
|
|
"DELETE FROM push_subscriptions WHERE endpoint = ?", (endpoint_index,)
|
|
)
|
|
connection.execute("DELETE FROM push_subscriptions WHERE session_id = ?", (session_id,))
|
|
connection.execute(
|
|
"INSERT INTO push_subscriptions(session_id, endpoint, subscription_json) VALUES (?, ?, ?)",
|
|
(session_id, endpoint_index, encoded),
|
|
)
|
|
|
|
def delete_session(self, session_id: str) -> None:
|
|
with self._connect() as connection:
|
|
connection.execute("DELETE FROM push_subscriptions WHERE session_id = ?", (session_id,))
|
|
|
|
def delete_all(self) -> None:
|
|
with self._connect() as connection:
|
|
connection.execute("DELETE FROM push_subscriptions")
|
|
|
|
def is_subscribed(self, session_id: str) -> bool:
|
|
with self._connect() as connection:
|
|
return connection.execute(
|
|
"SELECT 1 FROM push_subscriptions WHERE session_id = ?", (session_id,)
|
|
).fetchone() is not None
|
|
|
|
def subscription_for_session(self, session_id: str) -> dict | None:
|
|
with self._connect() as connection:
|
|
row = connection.execute(
|
|
"SELECT subscription_json FROM push_subscriptions WHERE session_id = ?",
|
|
(session_id,),
|
|
).fetchone()
|
|
return self._open_subscription(session_id, row[0]) if row else None
|
|
|
|
def delivery_health(self, session_id: str) -> dict:
|
|
with self._connect() as connection:
|
|
rows = connection.execute(
|
|
"""SELECT channel, consecutive_failures, last_attempted_at,
|
|
last_succeeded_at, reason
|
|
FROM push_delivery_health WHERE session_id = ? ORDER BY channel""",
|
|
(session_id,),
|
|
).fetchall()
|
|
return {
|
|
row[0]: {
|
|
"state": "degraded" if row[1] else "healthy",
|
|
"consecutive_failures": row[1],
|
|
"last_attempted_at": row[2],
|
|
"last_succeeded_at": row[3],
|
|
"reason": row[4],
|
|
}
|
|
for row in rows
|
|
}
|
|
|
|
def mark_delivery_failed(
|
|
self, session_id: str, channel: str, reason: str, *, now: float | None = None
|
|
) -> None:
|
|
attempted_at = time.time() if now is None else now
|
|
with self._connect() as connection:
|
|
connection.execute(
|
|
"""INSERT INTO push_delivery_health(
|
|
session_id, channel, consecutive_failures, last_attempted_at, reason
|
|
) VALUES (?, ?, 1, ?, ?)
|
|
ON CONFLICT(session_id, channel) DO UPDATE SET
|
|
consecutive_failures = consecutive_failures + 1,
|
|
last_attempted_at = excluded.last_attempted_at,
|
|
reason = excluded.reason""",
|
|
(session_id, channel, attempted_at, reason),
|
|
)
|
|
|
|
def mark_delivery_succeeded(
|
|
self, session_id: str, channel: str, *, now: float | None = None
|
|
) -> None:
|
|
attempted_at = time.time() if now is None else now
|
|
with self._connect() as connection:
|
|
connection.execute(
|
|
"""INSERT INTO push_delivery_health(
|
|
session_id, channel, consecutive_failures, last_attempted_at,
|
|
last_succeeded_at, reason
|
|
) VALUES (?, ?, 0, ?, ?, NULL)
|
|
ON CONFLICT(session_id, channel) DO UPDATE SET
|
|
consecutive_failures = 0,
|
|
last_attempted_at = excluded.last_attempted_at,
|
|
last_succeeded_at = excluded.last_succeeded_at,
|
|
reason = NULL""",
|
|
(session_id, channel, attempted_at, attempted_at),
|
|
)
|
|
|
|
def set_deadline_preferences(
|
|
self, session_id: str, *, enabled: bool, timezone: str, reminder_hour: int,
|
|
reminder_days: int = 2,
|
|
) -> None:
|
|
with self._connect() as connection:
|
|
connection.execute(
|
|
"""INSERT INTO push_deadline_preferences(
|
|
session_id, enabled, timezone, reminder_hour, reminder_days
|
|
) VALUES (?, ?, ?, ?, ?)
|
|
ON CONFLICT(session_id) DO UPDATE SET
|
|
enabled = excluded.enabled,
|
|
timezone = excluded.timezone,
|
|
reminder_hour = excluded.reminder_hour,
|
|
reminder_days = excluded.reminder_days""",
|
|
(session_id, int(enabled), timezone, reminder_hour, reminder_days),
|
|
)
|
|
|
|
def deadline_preferences(self, session_id: str, *, now: float | None = None) -> dict:
|
|
with self._connect() as connection:
|
|
row = connection.execute(
|
|
"""SELECT enabled, timezone, reminder_hour, reminder_days, snoozed_until
|
|
FROM push_deadline_preferences WHERE session_id = ?""",
|
|
(session_id,),
|
|
).fetchone()
|
|
current_time = time.time() if now is None else now
|
|
snoozed_until = row[4] if row and row[4] and row[4] > current_time else None
|
|
return {
|
|
"enabled": bool(row[0]) if row else False,
|
|
"timezone": row[1] if row else "UTC",
|
|
"reminder_hour": row[2] if row else 9,
|
|
"reminder_days": row[3] if row else 2,
|
|
"snoozed_until": snoozed_until,
|
|
}
|
|
|
|
def deadline_reminder_devices(self) -> list[DeadlineReminderDevice]:
|
|
with self._connect() as connection:
|
|
rows = connection.execute(
|
|
"""SELECT s.session_id, s.subscription_json, p.timezone,
|
|
p.reminder_hour, p.reminder_days, p.delivered_local_day,
|
|
p.snoozed_until
|
|
FROM push_subscriptions s
|
|
JOIN push_deadline_preferences p ON p.session_id = s.session_id
|
|
WHERE p.enabled = 1 ORDER BY s.session_id"""
|
|
).fetchall()
|
|
return [
|
|
DeadlineReminderDevice(
|
|
row[0],
|
|
self._open_subscription(row[0], row[1]),
|
|
row[2],
|
|
row[3],
|
|
row[4],
|
|
row[5],
|
|
row[6],
|
|
)
|
|
for row in rows
|
|
]
|
|
|
|
def snooze_deadline_reminder(
|
|
self, session_id: str, *, now: float, delay_seconds: int = 3_600
|
|
) -> bool:
|
|
with self._connect() as connection:
|
|
result = connection.execute(
|
|
"""UPDATE push_deadline_preferences SET snoozed_until = ?
|
|
WHERE session_id = ? AND enabled = 1""",
|
|
(now + delay_seconds, session_id),
|
|
)
|
|
return result.rowcount == 1
|
|
|
|
def clear_deadline_snooze(self, session_id: str) -> None:
|
|
with self._connect() as connection:
|
|
connection.execute(
|
|
"""UPDATE push_deadline_preferences SET snoozed_until = NULL
|
|
WHERE session_id = ?""",
|
|
(session_id,),
|
|
)
|
|
|
|
def mark_deadline_reminder_delivered(self, session_id: str, local_day: str) -> None:
|
|
with self._connect() as connection:
|
|
connection.execute(
|
|
"""UPDATE push_deadline_preferences
|
|
SET delivered_local_day = ?, snoozed_until = NULL
|
|
WHERE session_id = ? AND enabled = 1""",
|
|
(local_day, session_id),
|
|
)
|
|
|
|
def claim_unseen(
|
|
self, thread_revisions: Mapping[int, str] | Iterable[int | tuple[int, str]]
|
|
) -> list[PushDelivery]:
|
|
candidates = _revisions(thread_revisions)
|
|
if not candidates:
|
|
return []
|
|
with self._connect() as connection:
|
|
rows = connection.execute(
|
|
"SELECT session_id, subscription_json FROM push_subscriptions ORDER BY session_id"
|
|
).fetchall()
|
|
deliveries = []
|
|
for session_id, encoded in rows:
|
|
delivered = {
|
|
row[0]: row[1]
|
|
for row in connection.execute(
|
|
"SELECT thread_id, revision FROM push_deliveries WHERE session_id = ?",
|
|
(session_id,),
|
|
)
|
|
}
|
|
for thread_id, revision in candidates:
|
|
if delivered.get(thread_id) == "*":
|
|
connection.execute(
|
|
"""UPDATE push_deliveries SET revision = ?
|
|
WHERE session_id = ? AND thread_id = ?""",
|
|
(revision, session_id, thread_id),
|
|
)
|
|
delivered[thread_id] = revision
|
|
unseen = tuple(
|
|
candidate
|
|
for candidate in candidates
|
|
if (
|
|
candidate[0] not in delivered
|
|
or (
|
|
bool(candidate[1])
|
|
and (
|
|
not delivered[candidate[0]]
|
|
or candidate[1] > delivered[candidate[0]]
|
|
)
|
|
)
|
|
)
|
|
)
|
|
if unseen:
|
|
pending = {
|
|
row[0]: row[1]
|
|
for row in connection.execute(
|
|
"SELECT thread_id, revision FROM push_digest_pending WHERE session_id = ?",
|
|
(session_id,),
|
|
)
|
|
}
|
|
digest_revisions = []
|
|
for thread_id, revision in unseen:
|
|
if pending.get(thread_id) in {revision, "*"}:
|
|
digest_revisions.append((thread_id, revision))
|
|
if pending[thread_id] == "*":
|
|
connection.execute(
|
|
"""UPDATE push_digest_pending SET revision = ?
|
|
WHERE session_id = ? AND thread_id = ?""",
|
|
(revision, session_id, thread_id),
|
|
)
|
|
deliveries.append(
|
|
PushDelivery(
|
|
session_id,
|
|
self._open_subscription(session_id, encoded),
|
|
unseen,
|
|
tuple(digest_revisions),
|
|
)
|
|
)
|
|
return deliveries
|
|
|
|
def _open_subscription(self, session_id: str, payload: str) -> dict:
|
|
subscription, _plaintext = self._cipher.open(payload, binding=session_id)
|
|
if not isinstance(subscription, dict):
|
|
raise ValueError("push subscription payload is invalid")
|
|
return subscription
|
|
|
|
def reconcile_unread(self, thread_ids: Iterable[int]) -> None:
|
|
"""Prune per-device checkpoints that are absent from a complete snapshot."""
|
|
unread_ids = tuple(sorted({int(value) for value in thread_ids if int(value) > 0}))
|
|
with self._connect() as connection:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
connection.execute(
|
|
"CREATE TEMP TABLE current_unread(thread_id INTEGER PRIMARY KEY)"
|
|
)
|
|
connection.executemany(
|
|
"INSERT INTO current_unread(thread_id) VALUES (?)",
|
|
((thread_id,) for thread_id in unread_ids),
|
|
)
|
|
connection.execute(
|
|
"""DELETE FROM push_deliveries
|
|
WHERE NOT EXISTS (
|
|
SELECT 1 FROM current_unread
|
|
WHERE current_unread.thread_id = push_deliveries.thread_id
|
|
)"""
|
|
)
|
|
connection.execute(
|
|
"""DELETE FROM push_digest_pending
|
|
WHERE NOT EXISTS (
|
|
SELECT 1 FROM current_unread
|
|
WHERE current_unread.thread_id = push_digest_pending.thread_id
|
|
)"""
|
|
)
|
|
|
|
def mark_digest_pending(
|
|
self,
|
|
session_id: str,
|
|
thread_revisions: Mapping[int, str] | Iterable[int | tuple[int, str]],
|
|
) -> None:
|
|
values = _revisions(thread_revisions)
|
|
with self._connect() as connection:
|
|
connection.executemany(
|
|
"""INSERT INTO push_digest_pending(session_id, thread_id, revision)
|
|
VALUES (?, ?, ?)
|
|
ON CONFLICT(session_id, thread_id) DO UPDATE SET
|
|
revision = excluded.revision""",
|
|
((session_id, thread_id, revision) for thread_id, revision in values),
|
|
)
|
|
|
|
def mark_delivered(
|
|
self,
|
|
session_id: str,
|
|
thread_revisions: Mapping[int, str] | Iterable[int | tuple[int, str]],
|
|
) -> None:
|
|
values = _revisions(thread_revisions)
|
|
with self._connect() as connection:
|
|
connection.executemany(
|
|
"""INSERT INTO push_deliveries(session_id, thread_id, revision)
|
|
VALUES (?, ?, ?)
|
|
ON CONFLICT(session_id, thread_id) DO UPDATE SET
|
|
revision = excluded.revision""",
|
|
((session_id, thread_id, revision) for thread_id, revision in values),
|
|
)
|
|
connection.executemany(
|
|
"DELETE FROM push_digest_pending WHERE session_id = ? AND thread_id = ?",
|
|
((session_id, thread_id) for thread_id, _revision in values),
|
|
)
|