import 'dart:async'; import 'dart:typed_data'; import 'package:commons_core/commons_core.dart'; import 'package:equatable/equatable.dart'; import 'package:flutter_bloc/flutter_bloc.dart'; import '../data/variety_repository.dart'; import '../services/offer_mapper.dart'; import '../services/offer_outbox.dart'; import '../services/offer_thumbnail.dart'; import '../services/social_connection.dart'; import '../services/social_service.dart'; /// State of the offer discovery/publish screen. Transport-agnostic — it holds /// only what the UI shows, never a relay handle. class OffersState extends Equatable { const OffersState({ this.offers = const [], this.areaGeohash = '', this.query = '', this.typeFilter = const {}, this.categoryFilter = const {}, this.organicOnly = false, this.blockedAuthors = const {}, this.hiddenOfferKeys = const {}, this.searching = false, this.loadingMore = false, this.nextPageCursor, this.publishing = false, this.hasSearched = false, this.connectionEpoch = 0, this.error, }); /// Offers discovered so far for [areaGeohash], newest first. Bounded in size /// (see [OffersCubit.maxOffersKept]) so a busy area can't grow the list — and /// the inline photo thumbnails it carries — without limit. final List offers; /// The coarse area currently being browsed. final String areaGeohash; /// Free-text filter over [offers] typed in the search box (empty = show all). final String query; /// Active reciprocity-mode filter (gift/exchange/sale/wanted); empty = all. final Set typeFilter; /// Active category filter (free-text categories); empty = all. final Set categoryFilter; /// When true, show only offers the grower declared organic ("eco"). final bool organicOnly; /// Authors this user has blocked; their offers never surface. Kept in the /// state (not dropped at merge time) so unblocking can resurface what was /// already discovered. final Set blockedAuthors; /// Individual offers this user reported and hid ("author:id" keys, see /// SocialSettings.offerKey). Same idea as [blockedAuthors], finer grain. final Set hiddenOfferKeys; final bool searching; /// True while a "load more" page is in flight, so the UI shows a footer /// spinner and doesn't fire overlapping page requests. final bool loadingMore; /// Cursor (Unix seconds) for the next older page, or null when the newest /// page was short (nothing older) or the in-memory cap was reached — in both /// cases there is no more to load. final int? nextPageCursor; final bool publishing; /// Whether more offers can be paged in (drives the infinite-scroll trigger). bool get canLoadMore => nextPageCursor != null; /// True once a discovery has been started, so the UI can tell "no search yet" /// from "searched, found nothing". final bool hasSearched; /// Bumped whenever the underlying transport comes or goes, so the UI rebuilds /// and re-reads [OffersCubit.isOnline] — states differing only here are /// otherwise Equatable-equal and bloc would skip the emit. final int connectionEpoch; /// Last error, in human terms for the UI (null when fine). final String? error; /// [offers] narrowed by the text [query] and the chip filters (all ANDed). /// Kept separate from [offers] so filtering never drops the discoveries. List get visibleOffers { return offers.where((o) { // Moderation state (block/hide) is app-side, not part of the search facets. if (blockedAuthors.contains(o.authorPubkeyHex)) return false; if (hiddenOfferKeys.contains('${o.authorPubkeyHex}:${o.id}')) { return false; } // The query/facet chain is shared with saved-search alerts so the two // never disagree on what "matches". return SavedSearch.matchesFilters( o, query: query, types: typeFilter, categories: categoryFilter, organicOnly: organicOnly, ); }).toList(); } /// The distinct categories present across discovered [offers], sorted, so the /// UI only offers a category chip when some offer carries it. List get categories { final set = { for (final o in offers) if (o.category != null && o.category!.isNotEmpty) o.category!, }; final list = set.toList()..sort(); return list; } /// Whether any discovered offer declares organic, so the eco chip only shows /// when it can match something. bool get hasOrganic => offers.any((o) => o.isOrganic); bool get hasActiveFilter => typeFilter.isNotEmpty || categoryFilter.isNotEmpty || organicOnly; OffersState copyWith({ List? offers, String? areaGeohash, String? query, Set? typeFilter, Set? categoryFilter, bool? organicOnly, Set? blockedAuthors, Set? hiddenOfferKeys, bool? searching, bool? loadingMore, int? Function()? nextPageCursor, bool? publishing, bool? hasSearched, int? connectionEpoch, String? Function()? error, }) { return OffersState( offers: offers ?? this.offers, areaGeohash: areaGeohash ?? this.areaGeohash, query: query ?? this.query, typeFilter: typeFilter ?? this.typeFilter, categoryFilter: categoryFilter ?? this.categoryFilter, organicOnly: organicOnly ?? this.organicOnly, blockedAuthors: blockedAuthors ?? this.blockedAuthors, hiddenOfferKeys: hiddenOfferKeys ?? this.hiddenOfferKeys, searching: searching ?? this.searching, loadingMore: loadingMore ?? this.loadingMore, nextPageCursor: nextPageCursor != null ? nextPageCursor() : this.nextPageCursor, publishing: publishing ?? this.publishing, hasSearched: hasSearched ?? this.hasSearched, connectionEpoch: connectionEpoch ?? this.connectionEpoch, error: error != null ? error() : this.error, ); } @override List get props => [ offers, areaGeohash, query, typeFilter, categoryFilter, organicOnly, blockedAuthors, hiddenOfferKeys, searching, loadingMore, nextPageCursor, publishing, hasSearched, connectionEpoch, error, ]; } /// Drives offer discovery and publishing over an [OfferTransport]. Depends on /// the interface, not the Nostr backend, so it unit-tests with a fake and the /// UI stays offline-tolerant (a null transport = the social layer is unavailable /// and the screen degrades gracefully). class OffersCubit extends Cubit { OffersCubit( this._transport, { SocialConnection? connection, Future Function(String varietyId)? coverPhoto, String? Function(Uint8List bytes)? thumbnail, Future Function()? onDispose, }) : _coverPhoto = coverPhoto, _thumbnail = thumbnail, _onDispose = onDispose, super(const OffersState()) { // Fresh-install fix: the transport captured at build time may be null while // the shared connection is still coming up (or dropped). Follow the // connection so the market recovers by itself instead of staying offline // until a manual retry. _connSub = connection?.sessions.listen(_onSession); } OfferTransport? _transport; StreamSubscription? _connSub; /// The last area prefix asked of [discover]; re-run when the connection /// (re)appears so results show without the user tapping anything. String? _lastPrefix; void _onSession(SocialSession? session) { if (isClosed) return; final transport = session?.offers; if (identical(transport, _transport)) return; _transport = transport; final prefix = _lastPrefix; if (transport != null && prefix != null) { unawaited(discover(prefix)); // emits fresh states as results arrive } else { emit(state.copyWith(connectionEpoch: state.connectionEpoch + 1)); } } /// Fetches a variety's cover photo bytes; null in tests or when no inventory /// repo is wired. final Future Function(String varietyId)? _coverPhoto; /// Turns full photo bytes into a small `data:` thumbnail embedded in the offer /// (no media server). Null → offers publish without a photo. final String? Function(Uint8List bytes)? _thumbnail; /// Closes the owning [SocialSession]/connection when the cubit is disposed /// (null in tests, where the transport is a fake with nothing to close). final Future Function()? _onDispose; StreamSubscription? _sub; Timer? _searchTimeout; /// Upper bound on offers held in memory. Each offer can carry an inline /// (~40 KB) photo thumbnail, so an unbounded list in a busy area is a real /// out-of-memory risk on mobile. Paging stops once the list reaches this, and /// live offers beyond it evict the oldest — the newest stay visible. static const int maxOffersKept = 400; /// Whether a live transport is available (relay configured and reachable). bool get isOnline => _transport != null; /// Current time in Unix seconds — the granularity Nostr filters use for /// `since`/`until`. int _now() => DateTime.now().millisecondsSinceEpoch ~/ 1000; /// Starts (or restarts) discovery for [geohashPrefix]. Fetches the newest page /// up front (bounded), then keeps a live subscription for offers published /// from now on — older results arrive via [loadNextPage], not by draining the /// whole area into memory. Future discover(String geohashPrefix) async { _lastPrefix = geohashPrefix; final transport = _transport; if (transport == null) { emit(state.copyWith(error: () => 'offline', hasSearched: true)); return; } await _sub?.cancel(); _searchTimeout?.cancel(); emit(OffersState( areaGeohash: geohashPrefix, // Keep the text and chip filters across a refresh (only the results reset). query: state.query, typeFilter: state.typeFilter, categoryFilter: state.categoryFilter, organicOnly: state.organicOnly, blockedAuthors: state.blockedAuthors, hiddenOfferKeys: state.hiddenOfferKeys, searching: true, hasSearched: true, )); final query = DiscoveryQuery( geohashPrefix: geohashPrefix, types: state.typeFilter, ); // Live subscription first, bounded to offers published from now on, so a new // listing shows up immediately without re-dumping the whole area. final since = _now(); _sub = transport.discover(query, since: since).listen( (offer) => emit(state.copyWith( offers: _capped(_prepend(state.offers, offer)), searching: false, )), onError: (Object e) => emit(state.copyWith(searching: false, error: () => '$e')), ); try { final page = await transport.discoverPage(query); if (!isClosed) { emit(state.copyWith( offers: _capped(_mergePage(state.offers, page.offers)), nextPageCursor: () => page.nextCursor, searching: false, )); } } catch (e) { if (!isClosed) emit(state.copyWith(searching: false, error: () => '$e')); } // Safety net: if the first page hangs and no live offer arrives, still stop // the spinner so the screen shows the empty state, not an endless search. _searchTimeout = Timer(const Duration(seconds: 6), () { if (!isClosed && state.searching) { emit(state.copyWith(searching: false)); } }); } /// Loads the next (older) page of offers for the current area. Called by the /// list as it nears the end. No-op when offline, already loading, or there is /// nothing older to fetch. Future loadNextPage() async { final transport = _transport; final cursor = state.nextPageCursor; if (transport == null || cursor == null || state.loadingMore) return; emit(state.copyWith(loadingMore: true)); try { final page = await transport.discoverPage(DiscoveryQuery( geohashPrefix: state.areaGeohash, types: state.typeFilter, until: cursor, )); final merged = _mergePage(state.offers, page.offers); // Stop paging once the cap is reached — the list is already as large as we // keep — otherwise carry the transport's cursor onward. final reachedCap = merged.length >= maxOffersKept; emit(state.copyWith( offers: _capped(merged), nextPageCursor: () => reachedCap ? null : page.nextCursor, loadingMore: false, )); } catch (e) { emit(state.copyWith(loadingMore: false, error: () => '$e')); } } /// The offer's identity for de-duplication: NIP-99 addressable events are keyed /// by (author, `d`-tag id), so the same listing from two relays — or a stored /// copy plus a live echo — collapses to one entry. static String _key(Offer o) => '${o.authorPubkeyHex}:${o.id}'; /// Prepends a freshly-arrived (newer) [incoming] offer, dropping any existing /// entry with the same identity so a live echo never doubles the listing. static List _prepend(List current, Offer incoming) => [ incoming, for (final o in current) if (_key(o) != _key(incoming)) o, ]; /// Appends an older [page] after [current] (older offers sort below newer), /// skipping any already present. Keeps the newest-first ordering. static List _mergePage(List current, List page) { final seen = {for (final o in current) _key(o)}; return [ ...current, for (final o in page) if (seen.add(_key(o))) o, ]; } /// Caps the list to [maxOffersKept], keeping the newest (front) and dropping /// the oldest (tail) — bounds memory in a busy area. static List _capped(List offers) => offers.length <= maxOffersKept ? offers : offers.sublist(0, maxOffersKept); /// Narrows the visible offers to those whose summary matches [query]. Purely /// local over the already-discovered list; does not re-hit the transport. void search(String query) => emit(state.copyWith(query: query)); /// Toggles a reciprocity-mode chip on/off. Purely local over the discovered /// list; does not re-hit the transport. void toggleType(OfferType type) { final next = {...state.typeFilter}; next.contains(type) ? next.remove(type) : next.add(type); emit(state.copyWith(typeFilter: next)); } /// Toggles a category chip on/off. void toggleCategory(String category) { final next = {...state.categoryFilter}; next.contains(category) ? next.remove(category) : next.add(category); emit(state.copyWith(categoryFilter: next)); } /// Toggles the organic ("eco") filter. void toggleOrganicOnly() => emit(state.copyWith(organicOnly: !state.organicOnly)); /// Replaces the set of blocked authors (loaded from the local blocklist); /// their offers disappear from the visible list at once. void setBlockedAuthors(Set pubkeys) => emit(state.copyWith(blockedAuthors: pubkeys)); /// Replaces the set of locally hidden (reported) offers. void setHiddenOffers(Set offerKeys) => emit(state.copyWith(hiddenOfferKeys: offerKeys)); /// Applies a saved search's query and facets to the current discoveries. /// Purely local (does not re-hit the transport) — the area is unchanged, so /// the same discovered offers are simply re-narrowed by the saved filters. void applySavedSearch(SavedSearch search) => emit(state.copyWith( query: search.query, typeFilter: search.types, categoryFilter: search.categories, organicOnly: search.organicOnly, )); /// Clears every chip filter (leaves the text search untouched). void clearFilters() => emit(state.copyWith( typeFilter: const {}, categoryFilter: const {}, organicOnly: false, )); /// Publishes [offer]; returns the transport's verdict. No-op result when /// offline. Future publish(Offer offer) async { final transport = _transport; if (transport == null) { return const PublishResult(accepted: false, transportRef: '', message: 'offline'); } emit(state.copyWith(publishing: true, error: () => null)); try { final result = await transport.publish(offer); emit(state.copyWith( publishing: false, error: () => result.accepted ? null : result.message, )); return result; } catch (e) { emit(state.copyWith(publishing: false, error: () => '$e')); rethrow; } } /// Publishes the user's [lots] as offers, each tagged with the coarse /// [areaGeohash] and signed by [authorPubkeyHex]. Returns how many the relay /// accepted. No-op (returns 0) when offline or with no area set. Future publishLots( List lots, { required String authorPubkeyHex, required String areaGeohash, }) async { final transport = _transport; if (transport == null || areaGeohash.isEmpty || lots.isEmpty) return 0; emit(state.copyWith(publishing: true, error: () => null)); var accepted = 0; try { for (final lot in lots) { final offer = OfferMapper.fromSharedLot( lotId: lot.lotId, authorPubkeyHex: authorPubkeyHex, summary: lot.summary, sharing: lot.offerStatus, areaGeohash: areaGeohash, category: lot.category, isOrganic: lot.isOrganic, priceAmount: lot.priceAmount, priceCurrency: lot.priceCurrency, imageUrl: await _coverThumbnail(lot.varietyId), ); final result = await transport.publish(offer); if (result.accepted) accepted++; } emit(state.copyWith(publishing: false)); } catch (e) { emit(state.copyWith(publishing: false, error: () => '$e')); rethrow; } return accepted; } /// Builds a small inline `data:` thumbnail from the lot's cover photo to embed /// in the offer, or null when there's no photo, no thumbnailer, or it can't be /// shrunk to fit. Best-effort: a missing image never blocks publishing. Future _coverThumbnail(String varietyId) async { final coverPhoto = _coverPhoto; final thumbnail = _thumbnail; if (coverPhoto == null || thumbnail == null) return null; try { final Uint8List? bytes = await coverPhoto(varietyId); if (bytes == null || bytes.isEmpty) return null; return thumbnail(bytes); } catch (_) { return null; // degrade: publish the offer without a photo } } @override Future close() async { _searchTimeout?.cancel(); await _connSub?.cancel(); await _sub?.cancel(); await _onDispose?.call(); return super.close(); } } /// Opens an [OffersCubit] over the SHARED [SocialConnection], or an offline one /// that degrades gracefully. Local-first: when the connection isn't up (no relay /// configured / unreachable), the cubit gets a null transport and the screen /// still opens. Does NOT close the session — the connection owns it. Future createOffersCubit( SocialConnection connection, { VarietyRepository? repository, }) async { final session = await connection.session(); return OffersCubit( session?.offers, connection: connection, // keeps following (re)connects — see _onSession coverPhoto: repository?.coverPhotoFor, thumbnail: offerThumbnailDataUri, ); } /// Publishes any queued (offline-parked) lots now that we're online, then clears /// them from the [outbox]. Rebuilds each offer from the lot's CURRENT state, so /// since-deleted or now-private lots simply drop out. Returns how many published. Future flushOutbox({ required OfferOutbox outbox, required OffersCubit cubit, required List shareableLots, required String authorPubkeyHex, required String areaGeohash, }) async { if (!cubit.isOnline || areaGeohash.isEmpty) return 0; final queued = await outbox.pending(); if (queued.isEmpty) return 0; final toPublish = shareableLots.where((l) => queued.contains(l.lotId)).toList(); var published = 0; if (toPublish.isNotEmpty) { published = await cubit.publishLots( toPublish, authorPubkeyHex: authorPubkeyHex, areaGeohash: areaGeohash, ); } // Clear everything attempted (or gone) so we don't loop on it. await outbox.remove(queued); return published; }