peekExpired method

@override Future<Stream<String>> peekExpired({ DateTime? asOf, int? limit, })

Up to limit keys whose expiry instant is at-or-before asOf (default: now), in ascending-expiry order. limit: 1 answers "which key is next to expire". limit: null means no limit — the same set as getExpiredKeys, but with guaranteed ascending-expiry ordering. Ties (same expiry instant) are yielded in implementation-defined order.

Bounded counterpart to getExpiredKeys — use this for next-up probes and batched sweeps; the unbounded variant stays for the "drain everything" caller.

The returned Future completes once the backend has accepted the request; the Stream then yields the keys (empty stream if none match). Snapshot semantics as for nextExpiresAt.

Implementation

@override
Future<Stream<String>> peekExpired({DateTime? asOf, int? limit}) async {
  final cutoff = (asOf ?? DateTime.now()).toUtc().millisecondsSinceEpoch;
  var sql =
      'SELECT id FROM notifications WHERE expires_at IS NOT NULL AND expires_at <= ? ORDER BY expires_at';
  final params = <Object?>[cutoff];
  if (limit != null) {
    sql += ' LIMIT ?';
    params.add(limit);
  }
  final rows = _db.raw.select('$sql;', params);
  return Stream.fromIterable(rows.map((r) => r['id'] as String));
}