queryByPath method

@override Stream<KeyEntry<String, AtData, AtMetaData?>> queryByPath({ required KeyPattern keyPattern, required Predicate predicate, OrderByKey? orderBy, int? limit, int? skip, })

Stream every (key, value, metadata) entry matching keyPattern AND predicate. The predicate is evaluated against the value's JSON-shaped fields.

Backends MUST throw UnsupportedError when called and supportsPathQueries is false. Consumers should gate on the flag before calling.

Implementation

@override
Stream<KeyEntry<String, AtData, AtMetaData?>> queryByPath({
  required KeyPattern keyPattern,
  required Predicate predicate,
  OrderByKey? orderBy,
  int? limit,
  int? skip,
}) async* {
  final translated = SqliteQueryTranslator.predicateToSql(predicate, 'value');
  final where = '(${translated.sql})';
  final params = <Object?>[...translated.params];
  final order = switch (orderBy) {
    OrderByKey.byKey => ' ORDER BY atkey',
    OrderByKey.byExpiresAt => ' ORDER BY (expires_at IS NULL), expires_at',
    OrderByKey.byCreatedAt =>
      " ORDER BY json_extract(metadata, '\$.createdAt')",
    null => '',
  };
  final rows = _db.raw.select(
      'SELECT atkey, value, metadata FROM at_data WHERE $where$order;',
      params);
  var entries = rows
      .where((r) => SqliteQueryTranslator.matchesKeyPattern(
          r['atkey'] as String, keyPattern))
      .map((r) {
    final atkey = r['atkey'] as String;
    final atData = _atDataFromRow(r, atkey);
    return KeyEntry<String, AtData, AtMetaData?>(
        atkey, atData, atData.metaData);
  });
  if (skip != null) entries = entries.skip(skip);
  if (limit != null) entries = entries.take(limit);
  for (final e in entries) {
    yield e;
  }
}