@@ -137,26 +137,71 @@ async def dismiss_all_notifications(self) -> int:
137137 await self .db .commit ()
138138 return cursor .rowcount
139139
140- async def expire_notifications (self ) -> int :
140+ async def expire_due_notifications (self ) -> list [dict ]:
141+ """Flip pending rows past their expiry to ``expired``.
142+
143+ Returns the affected rows (as they were *before* the flip, with
144+ ``status`` already rewritten to ``'expired'`` in the returned
145+ dicts) so the service layer can report each expiry — inject a
146+ note into the asking session, audit-log approvals, gray the web
147+ card, edit the Telegram message. Select-then-update runs inside
148+ one transaction so a concurrent answer can't slip between the
149+ two statements.
150+ """
141151 now = datetime .now (timezone .utc ).isoformat ()
142- cursor = await self .db .execute (
143- """UPDATE notifications SET status = 'expired'
144- WHERE status = 'pending' AND expires_at IS NOT NULL AND expires_at < ?""" ,
145- (now ,),
146- )
147- await self .db .commit ()
148- return cursor .rowcount
152+ async with self ._atomic ():
153+ async with self .db .execute (
154+ """SELECT * FROM notifications
155+ WHERE status = 'pending'
156+ AND expires_at IS NOT NULL AND expires_at < ?""" ,
157+ (now ,),
158+ ) as cursor :
159+ rows = [dict (row ) async for row in cursor ]
160+ if rows :
161+ placeholders = "," .join ("?" for _ in rows )
162+ await self .db .execute (
163+ f"""UPDATE notifications SET status = 'expired'
164+ WHERE id IN ({ placeholders } ,
165+ tuple (r ["id" ] for r in rows ),
166+ )
167+ for r in rows :
168+ r ["status" ] = "expired"
169+ return rows
170+
171+ async def expire_notification (self , notification_id : str ) -> bool :
172+ """Flip a single pending row to ``expired``.
173+
174+ Used by the re-delivery tick when a row hits the
175+ ``max_redeliveries`` cap: instead of another fanout, it expires
176+ (with reporting) even though ``expires_at`` may still be in the
177+ future. Returns False if the row is not pending.
178+ """
179+ async with self ._atomic ():
180+ async with self .db .execute (
181+ "SELECT id FROM notifications WHERE id = ? AND status = 'pending'" ,
182+ (notification_id ,),
183+ ) as cursor :
184+ if not await cursor .fetchone ():
185+ return False
186+ await self .db .execute (
187+ "UPDATE notifications SET status = 'expired' WHERE id = ?" ,
188+ (notification_id ,),
189+ )
190+ return True
149191
150192 async def snooze_notification (
151- self , notification_id : str , new_expires_at : str ,
193+ self , notification_id : str , redeliver_at : str , new_expires_at : str ,
152194 ) -> bool :
153- """Push a pending notification's expiry forward .
195+ """Queue a pending notification for re-delivery .
154196
155- Used by the ``approval`` dispatcher when the user picks
156- ``snooze_24h``: the row stays at status=pending so a later
157- re-delivery tick (wired in PR 2) can surface it again, but the
158- expiry advances so it does not get caught by ``expire_stale``
159- in the meantime.
197+ Used when the user picks ``snooze_24h`` on an approval: the row
198+ stays at status=pending, ``redeliver_at`` marks when the
199+ periodic maintenance tick should fan it out again (fresh
200+ Telegram card + web broadcast), and ``expires_at`` advances past
201+ the re-delivery time so the row cannot expire before it
202+ resurfaces. Re-snoozing after a re-delivery simply sets
203+ ``redeliver_at`` again — each snooze buys another cycle, up to
204+ ``config.notifications.max_redeliveries``.
160205
161206 Returns True on success, False if the row is not pending.
162207 """
@@ -168,11 +213,64 @@ async def snooze_notification(
168213 if not await cursor .fetchone ():
169214 return False
170215 await self .db .execute (
171- "UPDATE notifications SET expires_at = ? WHERE id = ?" ,
172- (new_expires_at , notification_id ),
216+ """UPDATE notifications SET redeliver_at = ?, expires_at = ?
217+ WHERE id = ?""" ,
218+ (redeliver_at , new_expires_at , notification_id ),
173219 )
174220 return True
175221
222+ async def get_due_redeliveries (self ) -> list [dict ]:
223+ """Return pending rows whose ``redeliver_at`` has passed.
224+
225+ Oldest-first so long-waiting rows resurface before fresher ones
226+ when several come due in the same sweep.
227+ """
228+ now = datetime .now (timezone .utc ).isoformat ()
229+ async with self .db .execute (
230+ """SELECT * FROM notifications
231+ WHERE status = 'pending'
232+ AND redeliver_at IS NOT NULL AND redeliver_at <= ?
233+ ORDER BY created_at ASC""" ,
234+ (now ,),
235+ ) as cursor :
236+ return [dict (row ) async for row in cursor ]
237+
238+ async def mark_notification_redelivered (
239+ self , notification_id : str , new_expires_at : str | None = None ,
240+ ) -> bool :
241+ """Record one re-delivery: bump the count, clear ``redeliver_at``.
242+
243+ ``new_expires_at`` (when given) restarts the expiry window so
244+ the fresh card gets a full answering window — without it, a row
245+ whose original expiry already passed would be expired by the
246+ very next pass of the same sweep that just re-delivered it.
247+ Returns False if the row is not pending.
248+ """
249+ async with self ._atomic ():
250+ async with self .db .execute (
251+ "SELECT id FROM notifications WHERE id = ? AND status = 'pending'" ,
252+ (notification_id ,),
253+ ) as cursor :
254+ if not await cursor .fetchone ():
255+ return False
256+ if new_expires_at is not None :
257+ await self .db .execute (
258+ """UPDATE notifications
259+ SET redelivery_count = redelivery_count + 1,
260+ redeliver_at = NULL, expires_at = ?
261+ WHERE id = ?""" ,
262+ (new_expires_at , notification_id ),
263+ )
264+ else :
265+ await self .db .execute (
266+ """UPDATE notifications
267+ SET redelivery_count = redelivery_count + 1,
268+ redeliver_at = NULL
269+ WHERE id = ?""" ,
270+ (notification_id ,),
271+ )
272+ return True
273+
176274 async def count_pending_notifications (self , channel : str | None = None ) -> int :
177275 sql = "SELECT COUNT(*) FROM notifications WHERE status = 'pending'"
178276 params : tuple = ()
0 commit comments