feat(block2): Nostr transport backend in commons_core (all 3 contracts)

Second production slice: the concrete Nostr backend on the shared connection,
using the vetted nostr package's crypto (NIP-44 spec-vector tested, gift wrap).

- NostrConnection: one dart:io WebSocket + identity + EVENT/OK, REQ/EOSE via
  the library's Event/Filter/Request. Pure Dart, no Flutter.
- NostrOfferTransport (NIP-99 kind 30402) + Nip99Codec: geohash coarsened to a
  NIP-52 prefix ladder on the wire; Offer stays inventory/location-agnostic.
- NostrMessageTransport (NIP-17): send/inbox via the library's DirectMessage
  gift-wrap onion.
- NostrTrustTransport (custom addressable kind 30777): certify/revoke/discover.
- MiniRelay test support + integration tests: offers, messaging (metadata
  privacy), WoT membership + revoke, and trust-filters-offers — all over ONE
  shared connection, hermetic in-process relay.

Renamed the core offer lifecycle enum OfferStatus -> OfferLifecycle to avoid
colliding with app_seeds' sharing-intent OfferStatus (no Block 1 code touched;
app_seeds analyzes clean, 0 errors). commons_core: 55 tests green.
This commit is contained in:
vjrj 2026-07-10 02:38:50 +02:00
parent 1253f0c632
commit f50a4737cb
10 changed files with 791 additions and 3 deletions

View file

@ -13,6 +13,10 @@ export 'src/identity/nostr_key_derivation.dart';
export 'src/ids/id_gen.dart';
export 'src/social/certification.dart';
export 'src/social/message_transport.dart';
export 'src/social/nostr/nostr_connection.dart';
export 'src/social/nostr/nostr_message_transport.dart';
export 'src/social/nostr/nostr_offer_transport.dart';
export 'src/social/nostr/nostr_trust_transport.dart';
export 'src/social/offer.dart';
export 'src/social/offer_transport.dart';
export 'src/social/trust_transport.dart';

View file

@ -0,0 +1,97 @@
import 'package:nostr/nostr.dart';
import '../offer.dart';
/// Maps [Offer] Nostr NIP-99 "Classified Listing" events (kind 30402).
///
/// This codec enforces the privacy contract on the wire: the only location that
/// ever leaves the device is a coarse geohash, truncated here to
/// [maxGeohashChars] and emitted as a NIP-52 prefix ladder so area queries match
/// by exact tag. [Offer] itself has no field that could carry inventory or an
/// exact address the seam does the work.
class Nip99Codec {
/// NIP-99 addressable classified listing.
static const kindActive = 30402;
/// A 5-char geohash ±2.4 km cell the coarse "near you" of the mocks.
static const maxGeohashChars = 5;
/// Builds the signed tags + content for [offer]. Signing happens in the
/// transport via `Event.from`.
({List<List<String>> tags, String content}) encode(Offer offer) {
final coarse = _coarsen(offer.approxGeohash);
final tags = <List<String>>[
['d', offer.id],
['title', offer.summary],
['status', offer.status == OfferLifecycle.active ? 'active' : 'sold'],
['offer_type', offer.type.name],
];
for (var i = 1; i <= coarse.length; i++) {
tags.add(['g', coarse.substring(0, i)]); // NIP-52 prefix ladder, coarse only
}
if (offer.category != null) tags.add(['t', offer.category!]);
if (offer.radiusKm != null) tags.add(['radius_km', '${offer.radiusKm}']);
if (offer.type == OfferType.sale && offer.priceAmount != null) {
tags.add(['price', '${offer.priceAmount}', offer.priceCurrency ?? 'EUR']);
}
if (offer.type == OfferType.exchange && offer.exchangeTerms != null) {
tags.add(['exchange_terms', offer.exchangeTerms!]);
}
if (offer.imageUrl != null) tags.add(['image', offer.imageUrl!]);
if (offer.expiresAt != null) {
tags.add(['expiration', '${offer.expiresAt!.millisecondsSinceEpoch ~/ 1000}']);
}
return (tags: tags, content: offer.summary);
}
/// Reconstructs an [Offer] from a received event.
Offer decode(Event event) {
final price = _tag(event, 'price');
return Offer(
id: _tagValue(event, 'd') ?? event.id,
authorPubkeyHex: event.pubkey,
summary: _tagValue(event, 'title') ?? event.content,
type: _parseType(_tagValue(event, 'offer_type')),
status: _tagValue(event, 'status') == 'sold'
? OfferLifecycle.closed
: OfferLifecycle.active,
category: _tagValue(event, 't'),
approxGeohash: _longestGeohash(event),
radiusKm: int.tryParse(_tagValue(event, 'radius_km') ?? ''),
priceAmount: price != null && price.length > 1 ? num.tryParse(price[1]) : null,
priceCurrency: price != null && price.length > 2 ? price[2] : null,
exchangeTerms: _tagValue(event, 'exchange_terms'),
imageUrl: _tagValue(event, 'image'),
);
}
String _coarsen(String g) =>
g.length > maxGeohashChars ? g.substring(0, maxGeohashChars) : g;
String _longestGeohash(Event e) {
var best = '';
for (final t in e.tags) {
if (t.isNotEmpty && t[0] == 'g' && t.length > 1 && t[1].length > best.length) {
best = t[1];
}
}
return best;
}
String? _tagValue(Event e, String name) {
for (final t in e.tags) {
if (t.isNotEmpty && t[0] == name && t.length > 1) return t[1];
}
return null;
}
List<String>? _tag(Event e, String name) {
for (final t in e.tags) {
if (t.isNotEmpty && t[0] == name) return t;
}
return null;
}
OfferType _parseType(String? name) =>
OfferType.values.firstWhere((t) => t.name == name, orElse: () => OfferType.gift);
}

View file

@ -0,0 +1,107 @@
import 'dart:async';
import 'dart:convert';
import 'dart:io';
import 'package:nostr/nostr.dart';
import '../../identity/nostr_key_derivation.dart';
/// The ONE piece shared by every Nostr-backed transport: a single relay socket
/// plus the signing identity and the EVENT/OK, REQ/EOSE lifecycle.
/// [OfferTransport], [MessageTransport] and [TrustTransport] each sit on top of
/// this they share this *connection*, not a *contract*.
///
/// Pure Dart (`dart:io` WebSocket) no Flutter. A production relay *pool*
/// (multiple relays, reconnection, backoff) would wrap several of these; the
/// spike/foundation uses one relay.
class NostrConnection {
NostrConnection._(this._socket, this.identity) {
_socket.listen(_onData, onDone: _onDone, onError: (_) => _onDone());
}
final WebSocket _socket;
/// The derived identity this connection signs with.
final NostrIdentity identity;
String get publicKeyHex => identity.publicKeyHex;
String get privateKeyHex => identity.privateKeyHex;
final _incoming = StreamController<List<dynamic>>.broadcast();
int _subCounter = 0;
static Future<NostrConnection> connect(
String relayUrl, {
required NostrIdentity identity,
}) async {
final socket = await WebSocket.connect(relayUrl);
return NostrConnection._(socket, identity);
}
void _onData(dynamic d) {
if (!_incoming.isClosed) _incoming.add(jsonDecode(d as String) as List);
}
void _onDone() {
if (!_incoming.isClosed) _incoming.close();
}
/// Publishes an already-signed [event]; resolves with the relay's OK verdict.
Future<({bool accepted, String message})> publish(Event event) async {
final ok = _incoming.stream.firstWhere(
(m) => m[0] == 'OK' && m[1] == event.id,
);
_socket.add(jsonEncode(['EVENT', event.toMap()]));
final r = await ok.timeout(const Duration(seconds: 10));
return (accepted: r[2] as bool, message: r.length > 3 ? r[3] as String : '');
}
/// Streams events matching [filter] (live), until the caller cancels.
Stream<Event> subscribe(Filter filter) {
final subId = 'sub${_subCounter++}';
late StreamController<Event> controller;
StreamSubscription? sub;
controller = StreamController<Event>(
onListen: () {
sub = _incoming.stream.listen((m) {
if (m[0] == 'EVENT' && m[1] == subId) {
controller.add(
Event.fromMap(m[2] as Map<String, dynamic>, verify: false),
);
}
});
_socket.add(Request(subscriptionId: subId, filters: [filter]).serialize());
},
onCancel: () async {
_socket.add(jsonEncode(['CLOSE', subId]));
await sub?.cancel();
},
);
return controller.stream;
}
/// Collects stored events matching [filter] up to EOSE, then closes the sub.
Future<List<Event>> reqOnce(Filter filter) async {
final subId = 'once${_subCounter++}';
final results = <Event>[];
final done = Completer<void>();
final sub = _incoming.stream.listen((m) {
if (m[1] != subId) return;
if (m[0] == 'EVENT') {
results.add(Event.fromMap(m[2] as Map<String, dynamic>, verify: false));
} else if (m[0] == 'EOSE' && !done.isCompleted) {
done.complete();
}
});
_socket.add(Request(subscriptionId: subId, filters: [filter]).serialize());
await done.future.timeout(const Duration(seconds: 10));
_socket.add(jsonEncode(['CLOSE', subId]));
await sub.cancel();
return results;
}
Future<void> close() async {
await _socket.close();
if (!_incoming.isClosed) await _incoming.close();
}
}

View file

@ -0,0 +1,68 @@
import 'dart:async';
import 'package:nostr/nostr.dart' as nostr;
import '../message_transport.dart';
import 'nostr_connection.dart';
/// NIP-17 private-DM backend for [MessageTransport], on the shared
/// [NostrConnection]. The gift-wrap onion (rumor kind 14 seal kind 13 wrap
/// kind 1059 signed by an ephemeral key) and the NIP-44 encryption are done by
/// the vetted `nostr` package; we only route it over the relay.
class NostrMessageTransport implements MessageTransport {
NostrMessageTransport(this._conn);
final NostrConnection _conn;
static const kindGiftWrap = 1059;
@override
Future<void> send({required String toPubkey, required String text}) async {
final wrap = await nostr.DirectMessage.create(
message: text,
authorSecretKey: _conn.privateKeyHex,
recipientPubkey: toPubkey,
);
final r = await _conn.publish(wrap);
if (!r.accepted) throw StateError('relay rejected gift wrap: ${r.message}');
}
@override
Stream<PrivateMessage> inbox() => _conn
.subscribe(nostr.Filter(kinds: const [kindGiftWrap], pTags: [_conn.publicKeyHex]))
.asyncMap(_unwrap)
.where((m) => m != null)
.cast<PrivateMessage>();
/// Collect the inbox up to EOSE (tests/one-shot).
Future<List<PrivateMessage>> inboxUntilEose() async {
final wraps = await _conn.reqOnce(
nostr.Filter(kinds: const [kindGiftWrap], pTags: [_conn.publicKeyHex]),
);
final out = <PrivateMessage>[];
for (final w in wraps) {
final m = await _unwrap(w);
if (m != null) out.add(m);
}
return out;
}
Future<PrivateMessage?> _unwrap(nostr.Event wrap) async {
try {
final rumor = await nostr.DirectMessage.parse(
giftWrap: wrap,
recipientSecretKey: _conn.privateKeyHex,
);
return PrivateMessage(
fromPubkey: rumor.pubkey,
text: rumor.content,
at: DateTime.fromMillisecondsSinceEpoch(rumor.createdAt * 1000),
);
} catch (_) {
return null; // not for us / wrong key / tampered
}
}
@override
Future<void> close() => _conn.close();
}

View file

@ -0,0 +1,69 @@
import 'package:nostr/nostr.dart';
import '../offer.dart';
import '../offer_transport.dart';
import 'nip99_codec.dart';
import 'nostr_connection.dart';
/// Nostr NIP-99 backend for [OfferTransport], on a shared [NostrConnection].
class NostrOfferTransport implements OfferTransport {
NostrOfferTransport(this._conn);
final NostrConnection _conn;
final Nip99Codec _codec = Nip99Codec();
Filter _filter(DiscoveryQuery q) => Filter(
kinds: const [Nip99Codec.kindActive],
tagFilters: {'g': [q.geohashPrefix]},
limit: q.limit,
);
@override
Future<PublishResult> publish(Offer offer) async {
final encoded = _codec.encode(offer);
final event = Event.from(
kind: Nip99Codec.kindActive,
content: encoded.content,
tags: encoded.tags,
secretKey: _conn.privateKeyHex,
);
final r = await _conn.publish(event);
return PublishResult(
accepted: r.accepted,
transportRef: event.id,
message: r.message,
);
}
@override
Stream<Offer> discover(DiscoveryQuery query) => _conn
.subscribe(_filter(query))
.map(_codec.decode)
.where((o) => query.types.isEmpty || query.types.contains(o.type));
/// Collects matches up to EOSE (tests/one-shot browse).
Future<List<Offer>> discoverUntilEose(DiscoveryQuery query) async {
final events = await _conn.reqOnce(_filter(query));
return events
.map(_codec.decode)
.where((o) => query.types.isEmpty || query.types.contains(o.type))
.toList();
}
@override
Future<void> retract(String offerId) async {
// NIP-09 deletion of the addressable coordinate; the OK path mirrors publish.
final event = Event.from(
kind: 5,
content: 'offer retracted',
tags: [
['a', '${Nip99Codec.kindActive}:${_conn.publicKeyHex}:$offerId'],
],
secretKey: _conn.privateKeyHex,
);
await _conn.publish(event);
}
@override
Future<void> close() => _conn.close();
}

View file

@ -0,0 +1,97 @@
import 'package:nostr/nostr.dart';
import '../certification.dart';
import '../trust_transport.dart';
import 'nostr_connection.dart';
/// Duniter-style certifications carried as Nostr events, on the shared
/// [NostrConnection]. No settled NIP exists for a web of trust, so this uses a
/// custom addressable kind mapped to the Duniter model (g1-integration.md "WoT
/// propia pero compatible"): one live certification per (issuer, subject), so
/// re-certifying renews and revoking replaces.
class NostrTrustTransport implements TrustTransport {
NostrTrustTransport(this._conn);
final NostrConnection _conn;
/// Custom Tanemaki certification kind (addressable range).
static const kindCertification = 30777;
int get _now => DateTime.now().millisecondsSinceEpoch ~/ 1000;
@override
Future<void> certify({
required String subjectPubkey,
Duration validity = const Duration(days: 365),
String note = '',
}) =>
_publish(subjectPubkey, note: note, validity: validity, revoked: false);
@override
Future<void> revoke({required String subjectPubkey}) =>
_publish(subjectPubkey, note: '', validity: Duration.zero, revoked: true);
Future<void> _publish(
String subject, {
required String note,
required Duration validity,
required bool revoked,
}) async {
final event = Event.from(
kind: kindCertification,
content: note,
tags: [
['p', subject],
['d', subject],
['status', revoked ? 'revoked' : 'valid'],
if (!revoked) ['expiration', '${_now + validity.inSeconds}'],
],
secretKey: _conn.privateKeyHex,
);
final r = await _conn.publish(event);
if (!r.accepted) throw StateError('relay rejected cert: ${r.message}');
}
@override
Future<List<Certification>> allCertifications() async {
final events = await _conn.reqOnce(const Filter(kinds: [kindCertification]));
return events.map(_toCertification).toList();
}
@override
Future<Set<String>> certifiersOf(String subjectPubkey) async {
final events = await _conn.reqOnce(
Filter(kinds: const [kindCertification], pTags: [subjectPubkey]),
);
final now = DateTime.now();
return events
.map(_toCertification)
.where((c) => c.subject == subjectPubkey && c.isValidAt(now))
.map((c) => c.issuer)
.toSet();
}
Certification _toCertification(Event e) {
String? tag(String name) {
for (final t in e.tags) {
if (t.isNotEmpty && t[0] == name && t.length > 1) return t[1];
}
return null;
}
final exp = tag('expiration');
return Certification(
issuer: e.pubkey,
subject: tag('p') ?? tag('d') ?? '',
issuedAt: DateTime.fromMillisecondsSinceEpoch(e.createdAt * 1000),
expiresAt: exp == null
? null
: DateTime.fromMillisecondsSinceEpoch(int.parse(exp) * 1000),
revoked: tag('status') == 'revoked',
note: e.content,
);
}
@override
Future<void> close() => _conn.close();
}

View file

@ -2,7 +2,10 @@
/// reserves for a future "library of things".
enum OfferType { gift, exchange, sale, wanted, lend }
enum OfferStatus { active, reserved, closed }
/// The published offer's lifecycle (distinct from the domain's local
/// sharing-intent enum). Named `OfferLifecycle` to avoid colliding with
/// app-side `OfferStatus`.
enum OfferLifecycle { active, reserved, closed }
/// A publishable offer the "shop window", deliberately DECOUPLED from the
/// domain's inventory (sharing-model.md §2, core-domain-boundary.md §4.3).
@ -18,7 +21,7 @@ class Offer {
required this.summary,
required this.type,
required this.approxGeohash,
this.status = OfferStatus.active,
this.status = OfferLifecycle.active,
this.category,
this.priceAmount,
this.priceCurrency,
@ -38,7 +41,7 @@ class Offer {
final String summary;
final OfferType type;
final OfferStatus status;
final OfferLifecycle status;
/// Free-text category, prefilled in-app from the domain but opaque here.
final String? category;

View file

@ -0,0 +1,172 @@
import 'dart:convert';
import 'dart:typed_data';
import 'package:commons_core/commons_core.dart';
import 'package:test/test.dart';
import '../support/mini_relay.dart';
/// Integration: the three transports on one shared connection, over a hermetic
/// in-process relay, using the vetted `nostr` crypto. Mirrors the Block 2 spike
/// but on the production commons_core code.
void main() {
late MiniRelay relay;
setUp(() async => relay = await MiniRelay.start());
tearDown(() async => relay.stop());
Future<NostrIdentity> idFor(int fill) => NostrKeyDerivation.deriveFromSeed(
Uint8List(32)..fillRange(0, 32, fill),
);
Future<NostrConnection> connFor(NostrIdentity id) =>
NostrConnection.connect(relay.url, identity: id);
group('offers (NIP-99)', () {
test('published by one identity, discovered by another via geohash',
() async {
final alice = await idFor(1);
final bob = await idFor(2);
final aliceT = NostrOfferTransport(await connFor(alice));
final bobT = NostrOfferTransport(await connFor(bob));
final published = await aliceT.publish(Offer(
id: 'tomate-1',
authorPubkeyHex: alice.publicKeyHex,
summary: 'Tomate rosa de Barbastro',
type: OfferType.gift,
category: 'Solanaceae',
approxGeohash: 'sp3e9xyz',
));
expect(published.accepted, isTrue, reason: published.message);
final found = await bobT
.discoverUntilEose(const DiscoveryQuery(geohashPrefix: 'sp3'));
expect(found, hasLength(1));
expect(found.single.summary, 'Tomate rosa de Barbastro');
expect(found.single.authorPubkeyHex, alice.publicKeyHex);
await aliceT.close();
await bobT.close();
});
test('exact geohash never leaves the device (coarsened on the wire)',
() async {
final alice = await idFor(1);
final t = NostrOfferTransport(await connFor(alice));
await t.publish(Offer(
id: 'x',
authorPubkeyHex: alice.publicKeyHex,
summary: 'Pimiento',
type: OfferType.gift,
approxGeohash: 'sp3e9tunqmXY', // precise
));
final wire = jsonEncode(relay.eventsOfKind(30402));
expect(wire, isNot(contains('sp3e9tunqmXY')));
expect(wire, contains('sp3e9'));
await t.close();
});
});
group('messaging (NIP-17)', () {
test('Alice → Bob: arrives, sender authenticated, relay sees only cipher',
() async {
final alice = await idFor(1);
final bob = await idFor(2);
final aliceM = NostrMessageTransport(await connFor(alice));
final bobM = NostrMessageTransport(await connFor(bob));
await aliceM.send(
toPubkey: bob.publicKeyHex,
text: '¿cambiamos semilla de calabaza?',
);
final inbox = await bobM.inboxUntilEose();
expect(inbox, hasLength(1));
expect(inbox.single.text, '¿cambiamos semilla de calabaza?');
expect(inbox.single.fromPubkey, alice.publicKeyHex);
// Eavesdropper view: only kind-1059 ciphertext, ephemeral author, p=bob.
final wire = jsonEncode(relay.eventsOfKind(1059));
expect(wire, isNot(contains('calabaza')));
final wrap = relay.eventsOfKind(1059).single;
expect(wrap['pubkey'], isNot(alice.publicKeyHex));
await aliceM.close();
await bobM.close();
});
});
group('web of trust (custom kind)', () {
test('3 seeds certify a newcomer → known; revoke drops', () async {
final seeds = [for (var i = 1; i <= 3; i++) await idFor(i)];
final newcomer = await idFor(50);
for (final s in seeds) {
final t = NostrTrustTransport(await connFor(s));
await t.certify(subjectPubkey: newcomer.publicKeyHex);
await t.close();
}
final reader = NostrTrustTransport(await connFor(await idFor(99)));
expect(await reader.certifiersOf(newcomer.publicKeyHex), hasLength(3));
final wot = WebOfTrust.fromCertifications(
await reader.allCertifications(),
now: DateTime.now(),
);
final seedSet = seeds.map((s) => s.publicKeyHex).toSet();
expect(
wot.isMember(newcomer.publicKeyHex,
seeds: seedSet, threshold: 3, maxDistance: 5),
isTrue,
);
// One seed revokes certifier count drops.
final revoker = NostrTrustTransport(await connFor(seeds.first));
await revoker.revoke(subjectPubkey: newcomer.publicKeyHex);
expect(await reader.certifiersOf(newcomer.publicKeyHex), hasLength(2));
await reader.close();
await revoker.close();
});
test('trust filters offers: known author vs unknown stranger', () async {
final seeds = [for (var i = 1; i <= 3; i++) await idFor(i)];
final alice = await idFor(20);
final eve = await idFor(66);
for (final s in seeds) {
final t = NostrTrustTransport(await connFor(s));
await t.certify(subjectPubkey: alice.publicKeyHex);
await t.close();
}
for (final who in [alice, eve]) {
final o = NostrOfferTransport(await connFor(who));
await o.publish(Offer(
id: 'o-${who.publicKeyHex.substring(0, 6)}',
authorPubkeyHex: who.publicKeyHex,
summary: 'Semilla',
type: OfferType.gift,
approxGeohash: 'sp3e9',
));
await o.close();
}
// Bob: discover offers AND read trust over ONE shared connection.
final bobConn = await connFor(await idFor(7));
final found = await NostrOfferTransport(bobConn)
.discoverUntilEose(const DiscoveryQuery(geohashPrefix: 'sp3e9'));
final trust = NostrTrustTransport(bobConn);
final wot = WebOfTrust.fromCertifications(
await trust.allCertifications(),
now: DateTime.now(),
);
final seedSet = seeds.map((s) => s.publicKeyHex).toSet();
bool known(String pk) =>
wot.isMember(pk, seeds: seedSet, threshold: 3, maxDistance: 5);
expect(found, hasLength(2));
expect(known(alice.publicKeyHex), isTrue);
expect(known(eve.publicKeyHex), isFalse);
await trust.close();
});
});
}

View file

@ -0,0 +1,131 @@
import 'dart:async';
import 'dart:convert';
import 'dart:io';
import 'package:nostr/nostr.dart';
/// A tiny in-process Nostr relay (NIP-01 subset) for tests: EVENT / REQ / EOSE /
/// CLOSE, filtering by `kinds` / `authors` / `ids` / `#x` tags, with addressable
/// (3000039999) replacement by (kind, pubkey, `d`). Hermetic no network so
/// transport tests are CI-safe. Not a real relay.
class MiniRelay {
MiniRelay._(this._server);
final HttpServer _server;
final List<Map<String, dynamic>> _events = [];
final Set<_Sub> _subs = {};
int get port => _server.port;
String get url => 'ws://127.0.0.1:$port';
int get storedCount => _events.length;
/// All stored events of a given [kind] (for eavesdropper assertions).
List<Map<String, dynamic>> eventsOfKind(int kind) =>
_events.where((e) => e['kind'] == kind).toList();
static Future<MiniRelay> start() async {
final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
final relay = MiniRelay._(server);
relay._accept();
return relay;
}
void _accept() {
_server.listen((req) async {
if (!WebSocketTransformer.isUpgradeRequest(req)) {
req.response.statusCode = HttpStatus.badRequest;
await req.response.close();
return;
}
final ws = await WebSocketTransformer.upgrade(req);
final socketSubs = <_Sub>{};
ws.listen(
(data) => _onMessage(ws, socketSubs, data as String),
onDone: () => _subs.removeAll(socketSubs),
onError: (_) => _subs.removeAll(socketSubs),
);
});
}
void _onMessage(WebSocket ws, Set<_Sub> socketSubs, String data) {
final msg = jsonDecode(data) as List;
switch (msg[0]) {
case 'EVENT':
_handleEvent(ws, (msg[1] as Map).cast<String, dynamic>());
case 'REQ':
final subId = msg[1] as String;
final filters = msg.sublist(2).cast<Map<String, dynamic>>();
final sub = _Sub(ws, subId, filters);
_subs.add(sub);
socketSubs.add(sub);
for (final e in _events) {
if (sub.matches(e)) _send(ws, ['EVENT', subId, e]);
}
_send(ws, ['EOSE', subId]);
case 'CLOSE':
final subId = msg[1] as String;
socketSubs.removeWhere((s) => s.id == subId);
_subs.removeWhere((s) => s.ws == ws && s.id == subId);
}
}
void _handleEvent(WebSocket ws, Map<String, dynamic> event) {
final id = event['id'] as String;
try {
Event.fromMap(event, verify: true); // reject bad signatures like a relay
} catch (_) {
_send(ws, ['OK', id, false, 'invalid: bad signature']);
return;
}
final kind = event['kind'] as int;
if (kind >= 30000 && kind < 40000) {
final d = _tag(event, 'd') ?? '';
_events.removeWhere((e) =>
e['kind'] == kind && e['pubkey'] == event['pubkey'] && (_tag(e, 'd') ?? '') == d);
}
_events.add(event);
_send(ws, ['OK', id, true, '']);
for (final sub in _subs) {
if (sub.matches(event)) _send(sub.ws, ['EVENT', sub.id, event]);
}
}
void _send(WebSocket ws, Object message) => ws.add(jsonEncode(message));
Future<void> stop() => _server.close(force: true);
}
String? _tag(Map<String, dynamic> event, String name) {
for (final t in (event['tags'] as List)) {
final row = (t as List).cast<String>();
if (row.isNotEmpty && row[0] == name && row.length > 1) return row[1];
}
return null;
}
class _Sub {
_Sub(this.ws, this.id, this.filters);
final WebSocket ws;
final String id;
final List<Map<String, dynamic>> filters;
bool matches(Map<String, dynamic> e) => filters.any((f) => _matches(e, f));
bool _matches(Map<String, dynamic> e, Map<String, dynamic> f) {
if (f['kinds'] != null && !(f['kinds'] as List).contains(e['kind'])) return false;
if (f['authors'] != null && !(f['authors'] as List).contains(e['pubkey'])) return false;
if (f['ids'] != null && !(f['ids'] as List).contains(e['id'])) return false;
for (final entry in f.entries) {
if (!entry.key.startsWith('#')) continue;
final tagName = entry.key.substring(1);
final wanted = (entry.value as List).cast<String>();
final present = (e['tags'] as List)
.map((t) => (t as List).cast<String>())
.where((t) => t.isNotEmpty && t[0] == tagName && t.length > 1)
.map((t) => t[1]);
if (!wanted.any(present.contains)) return false;
}
return true;
}
}