peekNewlyAvailable method

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

Up to limit keys whose availableAt is in (since, asOf] — keys that crossed the born threshold since the caller's last sweep watermark. Ascending-availableAt order; asOf defaults to now.

The watermark pattern: born keys are persistent (unlike expired keys, which a sweep drains), so "all born keys" is useless as a trigger signal — every call would return the same set. Callers MUST track the largest availableAt they have already processed and pass it as since (exclusive). The first sweep may pass DateTime(0) or a persisted watermark. A key whose availableAt == since is NOT yielded; one whose availableAt == asOf IS.

The watermark is deliberately caller-side state: it adds no per-entry server state and composes with multiple independent readers, each holding its own watermark. If a future caller needs server-side "processed" tracking instead, revisit this contract before building around it.

Implementation

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