feat(mobile): websocket streaming with real-time messaging (#323)

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Wes
2026-04-14 20:20:35 -07:00
committed by GitHub
co-authored by Claude Opus 4.6
parent 0724f19382
commit 94d4e16461
28 changed files with 2111 additions and 36 deletions
@@ -59,6 +59,14 @@ pub fn sign_event(
Ok(event.as_json())
}
#[tauri::command]
pub fn get_nsec(state: State<'_, AppState>) -> Result<String, String> {
let keys = state.keys.lock().map_err(|error| error.to_string())?;
keys.secret_key()
.to_bech32()
.map_err(|error| format!("encode nsec: {error}"))
}
#[tauri::command]
pub fn create_auth_event(
challenge: String,
+1
View File
@@ -530,6 +530,7 @@ pub fn run() {
})
.invoke_handler(tauri::generate_handler![
get_identity,
get_nsec,
get_profile,
update_profile,
get_user_profile,
@@ -3,7 +3,7 @@ import { QRCodeSVG } from "qrcode.react";
import { Check, Copy, Loader2, Smartphone, TriangleAlert } from "lucide-react";
import { useMintTokenMutation } from "@/features/tokens/hooks";
import { getRelayHttpUrl } from "@/shared/api/tauri";
import { getNsec, getRelayHttpUrl } from "@/shared/api/tauri";
import type { TokenScope } from "@/shared/api/types";
import { Button } from "@/shared/ui/button";
import {
@@ -31,6 +31,7 @@ type PairingPayload = {
relayUrl: string;
token: string;
pubkey: string;
nsec: string;
};
function PairingDialog({
@@ -51,19 +52,21 @@ function PairingDialog({
const [copied, setCopied] = useState(false);
const generate = useCallback(async (pubkey: string) => {
const [tokenResult, relayUrl] = await Promise.all([
const [tokenResult, relayUrl, nsec] = await Promise.all([
mintRef.current({
name: `mobile-${Date.now()}`,
scopes: [...MOBILE_SCOPES],
expiresInDays: EXPIRES_IN_DAYS,
}),
getRelayHttpUrl(),
getNsec(),
]);
const payload: PairingPayload = {
relayUrl,
token: tokenResult.token,
pubkey,
nsec,
};
return `sprout://${toBase64Url(JSON.stringify(payload))}`;
}, []);
+4
View File
@@ -451,6 +451,10 @@ export async function getIdentity(): Promise<Identity> {
};
}
export async function getNsec(): Promise<string> {
return invokeTauri<string>("get_nsec");
}
export async function getProfile(): Promise<Profile> {
const profile = await invokeTauri<RawProfile>("get_profile");
return fromRawProfile(profile);
+6
View File
@@ -1,4 +1,6 @@
PODS:
- connectivity_plus (0.0.1):
- Flutter
- Flutter (1.0.0)
- flutter_secure_storage (6.0.0):
- Flutter
@@ -56,6 +58,7 @@ PODS:
- FlutterMacOS
DEPENDENCIES:
- connectivity_plus (from `.symlinks/plugins/connectivity_plus/ios`)
- Flutter (from `Flutter`)
- flutter_secure_storage (from `.symlinks/plugins/flutter_secure_storage/ios`)
- mobile_scanner (from `.symlinks/plugins/mobile_scanner/ios`)
@@ -76,6 +79,8 @@ SPEC REPOS:
- PromisesObjC
EXTERNAL SOURCES:
connectivity_plus:
:path: ".symlinks/plugins/connectivity_plus/ios"
Flutter:
:path: Flutter
flutter_secure_storage:
@@ -86,6 +91,7 @@ EXTERNAL SOURCES:
:path: ".symlinks/plugins/shared_preferences_foundation/darwin"
SPEC CHECKSUMS:
connectivity_plus: cb623214f4e1f6ef8fe7403d580fdad517d2f7dd
Flutter: cabc95a1d2626b1b06e7179b784ebcf0c0cde467
flutter_secure_storage: 1ed9476fba7e7a782b22888f956cce43e2c62f13
GoogleDataTransport: aae35b7ea0c09004c3797d53c8c41f66f219d6a7
+8
View File
@@ -5,6 +5,7 @@ import 'package:lucide_icons_flutter/lucide_icons.dart';
import 'features/home/home_page.dart';
import 'features/pairing/pairing_page.dart';
import 'shared/auth/auth.dart';
import 'shared/relay/relay.dart';
import 'shared/theme/theme.dart';
class App extends HookConsumerWidget {
@@ -19,6 +20,13 @@ class App extends HookConsumerWidget {
final lightScheme = applyAccent(lightColorScheme, accentIndex);
final darkScheme = applyAccent(darkColorScheme, accentIndex);
// Eagerly initialize websocket session and lifecycle observer when
// authenticated. These providers connect and manage the websocket.
if (authState.value?.status == AuthStatus.authenticated) {
ref.watch(relaySessionProvider);
ref.watch(appLifecycleProvider);
}
return MaterialApp(
title: 'Sprout',
theme: AppTheme.light(colorScheme: lightScheme),
+17
View File
@@ -61,4 +61,21 @@ class Channel {
bool get isForum => channelType == 'forum';
bool get isDm => channelType == 'dm';
bool get isPrivate => visibility == 'private';
Channel copyWith({DateTime? lastMessageAt}) => Channel(
id: id,
name: name,
channelType: channelType,
visibility: visibility,
description: description,
topic: topic,
purpose: purpose,
createdBy: createdBy,
createdAt: createdAt,
memberCount: memberCount,
lastMessageAt: lastMessageAt ?? this.lastMessageAt,
isMember: isMember,
ttlSeconds: ttlSeconds,
ttlDeadline: ttlDeadline,
);
}
@@ -0,0 +1,411 @@
import 'package:flutter/material.dart';
import 'package:flutter_hooks/flutter_hooks.dart';
import 'package:hooks_riverpod/hooks_riverpod.dart';
import 'package:lucide_icons_flutter/lucide_icons.dart';
import '../../shared/relay/relay.dart';
import '../../shared/theme/theme.dart';
import '../profile/user_cache_provider.dart';
import '../profile/user_profile.dart';
import 'channel.dart';
import 'channel_messages_provider.dart';
import 'channel_typing_provider.dart';
import 'send_message_provider.dart';
/// Fetch channel members and preload their profiles into the user cache.
Future<void> _preloadMembers(WidgetRef ref, String channelId) async {
try {
final client = ref.read(relayClientProvider);
final json =
await client.get('/api/channels/$channelId/members')
as Map<String, dynamic>;
final members = json['members'] as List<dynamic>? ?? [];
final pubkeys = members
.map((m) => (m as Map<String, dynamic>)['pubkey'] as String)
.toList();
if (pubkeys.isNotEmpty) {
ref.read(userCacheProvider.notifier).preload(pubkeys);
}
} catch (_) {
// Non-fatal — mentions will just fall back to cache from messages.
}
}
class ChannelDetailPage extends HookConsumerWidget {
final Channel channel;
const ChannelDetailPage({super.key, required this.channel});
@override
Widget build(BuildContext context, WidgetRef ref) {
final messagesState = ref.watch(channelMessagesProvider(channel.id));
final typingEntries = ref.watch(channelTypingProvider(channel.id));
// Preload channel member profiles so @mentions resolve correctly.
useEffect(() {
_preloadMembers(ref, channel.id);
return null;
}, [channel.id]);
return Scaffold(
appBar: AppBar(
title: Row(
children: [
Icon(
channel.isPrivate ? LucideIcons.lock : LucideIcons.hash,
size: 18,
color: context.colors.onSurfaceVariant,
),
const SizedBox(width: Grid.half),
Expanded(
child: Text(channel.name, overflow: TextOverflow.ellipsis),
),
],
),
),
body: Column(
children: [
Expanded(
child: messagesState.when(
loading: () => const Center(child: CircularProgressIndicator()),
error: (e, _) => Center(
child: Text(
'Failed to load messages',
style: context.textTheme.bodyMedium?.copyWith(
color: context.colors.error,
),
),
),
data: (messages) =>
_MessageList(messages: messages, channelId: channel.id),
),
),
if (typingEntries.isNotEmpty)
_TypingIndicator(entries: typingEntries),
_ComposeBar(channelId: channel.id),
],
),
);
}
}
class _MessageList extends ConsumerWidget {
final List<NostrEvent> messages;
final String channelId;
const _MessageList({required this.messages, required this.channelId});
@override
Widget build(BuildContext context, WidgetRef ref) {
if (messages.isEmpty) {
return Center(
child: Column(
mainAxisSize: MainAxisSize.min,
children: [
Icon(
LucideIcons.messageSquare,
size: Grid.xl,
color: context.colors.outline,
),
const SizedBox(height: Grid.xxs),
Text(
'No messages yet',
style: context.textTheme.bodyLarge?.copyWith(
color: context.colors.onSurfaceVariant,
),
),
const SizedBox(height: Grid.half),
Text(
'Be the first to say something!',
style: context.textTheme.bodySmall?.copyWith(
color: context.colors.outline,
),
),
],
),
);
}
// Filter to only renderable message events for display and grouping.
final displayMessages = messages
.where(
(e) =>
e.kind == EventKind.streamMessage ||
e.kind == EventKind.streamMessageV2,
)
.toList();
return ListView.builder(
reverse: true,
padding: const EdgeInsets.symmetric(
horizontal: Grid.xs,
vertical: Grid.xxs,
),
itemCount: displayMessages.length,
itemBuilder: (context, index) {
// Reversed list: index 0 = newest (bottom of screen).
final chronIdx = displayMessages.length - 1 - index;
final message = displayMessages[chronIdx];
// The message visually above is the one earlier in time.
final prevMessage = chronIdx > 0 ? displayMessages[chronIdx - 1] : null;
final showAuthor =
prevMessage == null ||
prevMessage.pubkey.toLowerCase() != message.pubkey.toLowerCase() ||
(message.createdAt - prevMessage.createdAt) > 300;
return _MessageBubble(event: message, showAuthor: showAuthor);
},
);
}
}
class _MessageBubble extends ConsumerWidget {
final NostrEvent event;
final bool showAuthor;
const _MessageBubble({required this.event, required this.showAuthor});
@override
Widget build(BuildContext context, WidgetRef ref) {
// Look up the user profile for display name.
final userCache = ref.watch(userCacheProvider);
final profile =
userCache[event.pubkey.toLowerCase()] ??
ref.read(userCacheProvider.notifier).get(event.pubkey.toLowerCase());
final displayName = profile?.label ?? _shortPubkey(event.pubkey);
return Padding(
padding: EdgeInsets.only(top: showAuthor ? Grid.xxs : Grid.quarter),
child: Row(
crossAxisAlignment: CrossAxisAlignment.start,
children: [
if (showAuthor)
_UserAvatar(profile: profile, pubkey: event.pubkey)
else
const SizedBox(width: 28),
const SizedBox(width: Grid.xxs),
Expanded(
child: Column(
crossAxisAlignment: CrossAxisAlignment.start,
children: [
if (showAuthor)
Padding(
padding: const EdgeInsets.only(bottom: Grid.quarter),
child: Row(
children: [
Text(
displayName,
style: context.textTheme.labelMedium?.copyWith(
fontWeight: FontWeight.w600,
color: context.colors.onSurface,
),
),
const SizedBox(width: Grid.xxs),
Text(
_formatTime(event.createdAt),
style: context.textTheme.labelSmall?.copyWith(
color: context.colors.outline,
),
),
],
),
),
Text(
event.content,
style: context.textTheme.bodyMedium?.copyWith(
color: context.colors.onSurface,
),
),
],
),
),
],
),
);
}
static String _shortPubkey(String pubkey) {
if (pubkey.length > 12) {
return '${pubkey.substring(0, 8)}…';
}
return pubkey;
}
static String _formatTime(int createdAt) {
final dt = DateTime.fromMillisecondsSinceEpoch(
createdAt * 1000,
isUtc: true,
).toLocal();
final now = DateTime.now();
final diff = now.difference(dt);
if (diff.inDays > 0) {
return '${dt.month}/${dt.day} ${_pad(dt.hour)}:${_pad(dt.minute)}';
}
return '${_pad(dt.hour)}:${_pad(dt.minute)}';
}
static String _pad(int n) => n.toString().padLeft(2, '0');
}
class _UserAvatar extends StatelessWidget {
final UserProfile? profile;
final String pubkey;
const _UserAvatar({required this.profile, required this.pubkey});
@override
Widget build(BuildContext context) {
final initial =
profile?.initial ?? (pubkey.isNotEmpty ? pubkey[0].toUpperCase() : '?');
final avatarUrl = profile?.avatarUrl;
return CircleAvatar(
radius: 14,
backgroundColor: context.colors.primaryContainer,
backgroundImage: avatarUrl != null ? NetworkImage(avatarUrl) : null,
child: avatarUrl == null
? Text(
initial,
style: context.textTheme.labelSmall?.copyWith(
color: context.colors.onPrimaryContainer,
fontWeight: FontWeight.w600,
),
)
: null,
);
}
}
class _TypingIndicator extends ConsumerWidget {
final List<TypingEntry> entries;
const _TypingIndicator({required this.entries});
@override
Widget build(BuildContext context, WidgetRef ref) {
final userCache = ref.watch(userCacheProvider);
final names = entries.map((e) {
final profile =
userCache[e.pubkey.toLowerCase()] ??
ref.read(userCacheProvider.notifier).get(e.pubkey.toLowerCase());
return profile?.label ?? _shortPubkey(e.pubkey);
}).toList();
final text = switch (names.length) {
1 => '${names[0]} is typing…',
2 => '${names[0]} and ${names[1]} are typing…',
_ => '${names[0]} and ${names.length - 1} others are typing…',
};
return Container(
width: double.infinity,
padding: const EdgeInsets.symmetric(
horizontal: Grid.xs,
vertical: Grid.quarter + 2,
),
child: Text(
text,
style: context.textTheme.labelSmall?.copyWith(
color: context.colors.outline,
fontStyle: FontStyle.italic,
),
),
);
}
static String _shortPubkey(String pubkey) {
if (pubkey.length > 12) return '${pubkey.substring(0, 8)}…';
return pubkey;
}
}
class _ComposeBar extends HookConsumerWidget {
final String channelId;
const _ComposeBar({required this.channelId});
@override
Widget build(BuildContext context, WidgetRef ref) {
final controller = useTextEditingController();
final isSending = useState(false);
Future<void> send() async {
final text = controller.text.trim();
if (text.isEmpty || isSending.value) return;
isSending.value = true;
try {
await ref
.read(sendMessageProvider)
.call(channelId: channelId, content: text);
controller.clear();
} finally {
isSending.value = false;
}
}
return Container(
decoration: BoxDecoration(
border: Border(top: BorderSide(color: context.colors.outlineVariant)),
color: context.colors.surface,
),
padding: EdgeInsets.only(
left: Grid.xs,
right: Grid.xxs,
top: Grid.xxs,
bottom: MediaQuery.viewPaddingOf(context).bottom + Grid.xxs,
),
child: Row(
children: [
Expanded(
child: TextField(
controller: controller,
textInputAction: TextInputAction.send,
onSubmitted: (_) => send(),
minLines: 1,
maxLines: 5,
decoration: InputDecoration(
hintText: 'Message…',
border: OutlineInputBorder(
borderRadius: BorderRadius.circular(Radii.lg),
borderSide: BorderSide(color: context.colors.outlineVariant),
),
enabledBorder: OutlineInputBorder(
borderRadius: BorderRadius.circular(Radii.lg),
borderSide: BorderSide(color: context.colors.outlineVariant),
),
focusedBorder: OutlineInputBorder(
borderRadius: BorderRadius.circular(Radii.lg),
borderSide: BorderSide(color: context.colors.primary),
),
contentPadding: const EdgeInsets.symmetric(
horizontal: Grid.twelve,
vertical: Grid.xxs,
),
isDense: true,
),
),
),
const SizedBox(width: Grid.half),
IconButton(
onPressed: isSending.value ? null : send,
icon: isSending.value
? SizedBox(
width: 20,
height: 20,
child: CircularProgressIndicator(
strokeWidth: 2,
color: context.colors.primary,
),
)
: Icon(
LucideIcons.sendHorizontal,
color: context.colors.primary,
),
),
],
),
);
}
}
@@ -0,0 +1,114 @@
import 'package:hooks_riverpod/hooks_riverpod.dart';
import '../../shared/relay/relay.dart';
/// Provides the message list for a specific channel. Fetches history on init,
/// then subscribes to live events via the websocket session.
class ChannelMessagesNotifier extends Notifier<AsyncValue<List<NostrEvent>>> {
final String channelId;
void Function()? _unsubscribe;
ChannelMessagesNotifier(this.channelId);
@override
AsyncValue<List<NostrEvent>> build() {
final sessionState = ref.watch(relaySessionProvider);
ref.onDispose(() {
_unsubscribe?.call();
_unsubscribe = null;
});
if (sessionState.status != SessionStatus.connected) {
return const AsyncData([]);
}
_init();
return const AsyncLoading();
}
Future<void> _init() async {
try {
final session = ref.read(relaySessionProvider.notifier);
// 1. Fetch recent history via REQ/EOSE.
final history = await session.fetchHistory(
NostrFilter(
kinds: EventKind.channelEventKinds,
tags: {
'#h': [channelId],
},
limit: 50,
),
);
// 2. Subscribe to live events for this channel.
_unsubscribe = await session.subscribe(
NostrFilter(
kinds: EventKind.channelEventKinds,
tags: {
'#h': [channelId],
},
limit: 0,
),
_handleLiveEvent,
);
history.sort((a, b) => a.createdAt.compareTo(b.createdAt));
state = AsyncData(history);
} catch (e, st) {
state = AsyncError(e, st);
}
}
void _handleLiveEvent(NostrEvent event) {
state = state.whenData((events) => _mergeEvent(events, event));
}
/// Merge a new event into the sorted list, deduplicating by ID.
static List<NostrEvent> _mergeEvent(
List<NostrEvent> current,
NostrEvent incoming,
) {
if (current.any((e) => e.id == incoming.id)) return current;
final updated = [...current, incoming];
updated.sort((a, b) => a.createdAt.compareTo(b.createdAt));
return updated;
}
/// Fetch older messages (pagination). Call this when the user scrolls up.
Future<void> fetchOlder() async {
final currentEvents = state.value;
if (currentEvents == null || currentEvents.isEmpty) return;
final oldest = currentEvents.first.createdAt;
final session = ref.read(relaySessionProvider.notifier);
final older = await session.fetchHistory(
NostrFilter(
kinds: EventKind.channelEventKinds,
tags: {
'#h': [channelId],
},
limit: 50,
until: oldest,
),
);
if (older.isEmpty) return;
state = state.whenData((events) {
final ids = events.map((e) => e.id).toSet();
final deduped = older.where((e) => !ids.contains(e.id)).toList();
final merged = [...deduped, ...events];
merged.sort((a, b) => a.createdAt.compareTo(b.createdAt));
return merged;
});
}
}
final channelMessagesProvider =
NotifierProvider.family<
ChannelMessagesNotifier,
AsyncValue<List<NostrEvent>>,
String
>(ChannelMessagesNotifier.new);
@@ -0,0 +1,113 @@
import 'dart:async';
import 'package:flutter/foundation.dart';
import 'package:hooks_riverpod/hooks_riverpod.dart';
import '../../shared/relay/relay.dart';
/// A single typing indicator entry.
@immutable
class TypingEntry {
final String pubkey;
final String? threadHeadId;
final int expiresAtMs;
const TypingEntry({
required this.pubkey,
this.threadHeadId,
required this.expiresAtMs,
});
}
/// Tracks who is currently typing in a specific channel.
///
/// Subscribes to kind:20002 (typing indicator) events via websocket.
/// Entries expire after 8 seconds (matching the desktop TTL).
class ChannelTypingNotifier extends Notifier<List<TypingEntry>> {
static const _ttlMs = 8000;
static const _pruneIntervalMs = 1000;
final String channelId;
void Function()? _unsubscribe;
Timer? _pruneTimer;
ChannelTypingNotifier(this.channelId);
@override
List<TypingEntry> build() {
final sessionState = ref.watch(relaySessionProvider);
ref.onDispose(() {
_unsubscribe?.call();
_unsubscribe = null;
_pruneTimer?.cancel();
_pruneTimer = null;
});
if (sessionState.status == SessionStatus.connected) {
_subscribeLive();
}
return [];
}
void _subscribeLive() async {
final session = ref.read(relaySessionProvider.notifier);
_unsubscribe = await session.subscribe(
NostrFilter(
kinds: [EventKind.typingIndicator],
tags: {
'#h': [channelId],
},
limit: 10,
),
_handleTypingEvent,
);
}
void _handleTypingEvent(NostrEvent event) {
final now = DateTime.now().millisecondsSinceEpoch;
final entry = TypingEntry(
pubkey: event.pubkey,
threadHeadId: event.getTagValue('e'),
expiresAtMs: now + _ttlMs,
);
// Upsert: replace existing entry for same pubkey+thread, or add.
final updated =
state
.where(
(e) =>
!(e.pubkey == entry.pubkey &&
e.threadHeadId == entry.threadHeadId),
)
.toList()
..add(entry);
state = updated;
_ensurePruneTimer();
}
void _ensurePruneTimer() {
_pruneTimer ??= Timer.periodic(
const Duration(milliseconds: _pruneIntervalMs),
(_) => _prune(),
);
}
void _prune() {
final now = DateTime.now().millisecondsSinceEpoch;
final pruned = state.where((e) => e.expiresAtMs > now).toList();
state = pruned;
if (pruned.isEmpty) {
_pruneTimer?.cancel();
_pruneTimer = null;
}
}
}
final channelTypingProvider =
NotifierProvider.family<ChannelTypingNotifier, List<TypingEntry>, String>(
ChannelTypingNotifier.new,
);
+70 -21
View File
@@ -1,45 +1,43 @@
import 'dart:async';
import 'dart:io';
import 'package:flutter/material.dart';
import 'package:flutter_hooks/flutter_hooks.dart';
import 'package:hooks_riverpod/hooks_riverpod.dart';
import 'package:lucide_icons_flutter/lucide_icons.dart';
import '../../shared/relay/relay_client.dart';
import '../../shared/relay/relay.dart';
import '../../shared/theme/theme.dart';
import '../profile/profile_avatar.dart';
import 'channel.dart';
import 'channel_detail_page.dart';
import 'channels_provider.dart';
class ChannelsPage extends HookConsumerWidget {
class ChannelsPage extends ConsumerWidget {
const ChannelsPage({super.key});
@override
Widget build(BuildContext context, WidgetRef ref) {
final channelsAsync = ref.watch(channelsProvider);
// Poll every 30s while this page is mounted, matching desktop's pattern.
useEffect(() {
final timer = Timer.periodic(
const Duration(seconds: 30),
(_) => ref.read(channelsProvider.notifier).refresh(),
);
return timer.cancel;
}, const []);
final sessionState = ref.watch(relaySessionProvider);
return Scaffold(
appBar: AppBar(
title: const Text('Sprout'),
actions: const [ProfileAvatar()],
),
body: channelsAsync.when(
loading: () => const Center(child: CircularProgressIndicator()),
error: (error, _) => _ErrorView(
error: error,
onRetry: () => ref.read(channelsProvider.notifier).refresh(),
),
data: (channels) => _ChannelsList(channels: channels),
body: Column(
children: [
_ConnectionBanner(status: sessionState.status),
Expanded(
child: channelsAsync.when(
loading: () => const Center(child: CircularProgressIndicator()),
error: (error, _) => _ErrorView(
error: error,
onRetry: () => ref.read(channelsProvider.notifier).refresh(),
),
data: (channels) => _ChannelsList(channels: channels),
),
),
],
),
);
}
@@ -148,7 +146,11 @@ class _ChannelTile extends StatelessWidget {
return InkWell(
borderRadius: BorderRadius.circular(Radii.md),
onTap: () {
// TODO: navigate to channel
Navigator.of(context).push(
MaterialPageRoute<void>(
builder: (_) => ChannelDetailPage(channel: channel),
),
);
},
child: Padding(
padding: const EdgeInsets.symmetric(
@@ -279,6 +281,53 @@ class _EphemeralBadge extends StatelessWidget {
}
}
/// Slim banner shown when the websocket is reconnecting or disconnected.
class _ConnectionBanner extends StatelessWidget {
final SessionStatus status;
const _ConnectionBanner({required this.status});
@override
Widget build(BuildContext context) {
if (status == SessionStatus.connected ||
status == SessionStatus.disconnected) {
return const SizedBox.shrink();
}
final isConnecting = status == SessionStatus.connecting;
final message = isConnecting ? 'Connecting…' : 'Reconnecting…';
return Container(
width: double.infinity,
padding: const EdgeInsets.symmetric(
horizontal: Grid.xs,
vertical: Grid.quarter + 2,
),
color: context.colors.surfaceContainerHighest,
child: Row(
mainAxisAlignment: MainAxisAlignment.center,
children: [
SizedBox(
width: 12,
height: 12,
child: CircularProgressIndicator(
strokeWidth: 2,
color: context.colors.onSurfaceVariant,
),
),
const SizedBox(width: Grid.xxs),
Text(
message,
style: context.textTheme.labelSmall?.copyWith(
color: context.colors.onSurfaceVariant,
),
),
],
),
);
}
}
class _ErrorView extends StatelessWidget {
final Object error;
final VoidCallback onRetry;
@@ -4,11 +4,28 @@ import '../../shared/relay/relay.dart';
import 'channel.dart';
class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
void Function()? _unsubscribe;
@override
Future<List<Channel>> build() {
// Watch relayClientProvider here so we auto-refetch when config changes.
ref.watch(relayClientProvider);
return _fetch();
final sessionState = ref.watch(relaySessionProvider);
ref.onDispose(() {
_unsubscribe?.call();
_unsubscribe = null;
});
// Initial fetch via HTTP (reliable, paginated).
final fetchFuture = _fetch();
// If websocket is connected, subscribe to live channel events to keep
// the list up to date without polling.
if (sessionState.status == SessionStatus.connected) {
_subscribeLive();
}
return fetchFuture;
}
Future<List<Channel>> _fetch() async {
@@ -23,6 +40,40 @@ class ChannelsNotifier extends AsyncNotifier<List<Channel>> {
return channels;
}
void _subscribeLive() async {
final session = ref.read(relaySessionProvider.notifier);
_unsubscribe = await session.subscribe(
NostrFilter(kinds: EventKind.channelEventKinds, limit: 0),
_handleLiveEvent,
);
}
void _handleLiveEvent(NostrEvent event) {
final channelId = event.channelId;
if (channelId == null) return;
state = state.whenData((channels) {
final idx = channels.indexWhere((c) => c.id == channelId);
if (idx == -1) {
// Unknown channel — queue a full refresh to pick it up.
refresh();
return channels;
}
// Update lastMessageAt for the affected channel.
final updated = List<Channel>.of(channels);
final channel = updated[idx];
final eventTime = DateTime.fromMillisecondsSinceEpoch(
event.createdAt * 1000,
isUtc: true,
);
if (channel.lastMessageAt == null ||
eventTime.isAfter(channel.lastMessageAt!)) {
updated[idx] = channel.copyWith(lastMessageAt: eventTime);
}
return updated;
});
}
Future<void> refresh() async {
state = await AsyncValue.guard(_fetch);
}
@@ -0,0 +1,114 @@
import 'dart:convert';
import 'package:hooks_riverpod/hooks_riverpod.dart';
import 'package:nostr/nostr.dart' as nostr;
import '../../shared/relay/relay.dart';
import '../profile/user_cache_provider.dart';
import '../profile/user_profile.dart';
/// Sends messages via the relay HTTP API. The event is signed on the client
/// with the user's nsec, then POSTed as a full signed Nostr event — matching
/// what the desktop does via `submit_event`.
class SendMessage {
final RelayClient _client;
final String? _nsec;
final Map<String, UserProfile> Function() _readUserCache;
SendMessage({
required RelayClient client,
required String? nsec,
required Map<String, UserProfile> Function() readUserCache,
}) : _client = client,
_nsec = nsec,
_readUserCache = readUserCache;
/// Send a text message to a channel.
Future<void> call({
required String channelId,
required String content,
String? parentEventId,
List<String>? mentionPubkeys,
}) async {
final nsec = _nsec;
if (nsec == null || nsec.isEmpty) {
throw Exception('Cannot send messages: no signing key available');
}
// Decode bech32 nsec to hex private key.
final privkeyHex = nostr.Nip19.decodePrivkey(nsec);
if (privkeyHex.isEmpty) {
throw Exception('Invalid nsec');
}
// Resolve @mentions in the message content to pubkeys.
final resolvedMentions = mentionPubkeys ?? _resolveMentions(content);
final tags = <List<String>>[
['h', channelId],
if (parentEventId != null) ['e', parentEventId],
for (final pk in resolvedMentions) ['p', pk],
];
// Create and sign the event using the nostr package.
final event = nostr.Event.from(
kind: EventKind.streamMessage,
content: content,
tags: tags,
privkey: privkeyHex,
verify: false,
);
// POST the full signed event JSON to the relay.
final response = await _client.postRaw(
'/api/events',
body: jsonEncode(event.toJson()),
);
final result = jsonDecode(response) as Map<String, dynamic>;
if (result['accepted'] != true) {
throw Exception(result['message'] ?? 'Event rejected by relay');
}
}
/// Parse @mentions from message content and resolve to pubkeys using
/// the user cache. Reads the cache at call time (not construction time)
/// to ensure freshly loaded profiles are available.
List<String> _resolveMentions(String content) {
final mentionPattern = RegExp(r'@(\w+)');
final matches = mentionPattern.allMatches(content);
final pubkeys = <String>{};
// Read the current cache state at send time.
final cache = _readUserCache();
for (final match in matches) {
final name = match.group(1)?.toLowerCase();
if (name == null || name.isEmpty) continue;
for (final profile in cache.values) {
final displayName = profile.displayName?.toLowerCase();
if (displayName == null) continue;
// Match against full display name or first word of it.
final firstName = displayName.split(RegExp(r'\s+')).first;
if (displayName == name || firstName == name) {
pubkeys.add(profile.pubkey);
break;
}
}
}
return pubkeys.toList();
}
}
final sendMessageProvider = Provider<SendMessage>((ref) {
final client = ref.watch(relayClientProvider);
final config = ref.watch(relayConfigProvider);
return SendMessage(
client: client,
nsec: config.nsec,
readUserCache: () => ref.read(userCacheProvider),
);
});
@@ -109,6 +109,7 @@ class PairingNotifier extends Notifier<PairingState> {
relayUrl: relayUrl,
token: token,
pubkey: decoded['pubkey'] as String?,
nsec: decoded['nsec'] as String?,
);
}
@@ -1,3 +1,6 @@
import 'dart:async';
import 'package:flutter/widgets.dart';
import 'package:hooks_riverpod/hooks_riverpod.dart';
import '../../shared/relay/relay.dart';
@@ -33,14 +36,56 @@ final profileProvider = AsyncNotifierProvider<ProfileNotifier, UserProfile?>(
);
/// Presence status for the current user.
///
/// Sends a heartbeat every 60s while the app is active. Watches
/// [appLifecycleProvider] to send "away" when backgrounded.
class PresenceNotifier extends AsyncNotifier<String> {
static const _heartbeatInterval = Duration(seconds: 60);
Timer? _heartbeatTimer;
@override
Future<String> build() {
ref.watch(relayClientProvider);
ref.watch(profileProvider);
final lifecycle = ref.watch(appLifecycleProvider);
ref.onDispose(() {
_heartbeatTimer?.cancel();
_heartbeatTimer = null;
});
if (lifecycle == AppLifecycleState.resumed) {
_startHeartbeat();
return _setPresence('online');
} else if (lifecycle == AppLifecycleState.paused ||
lifecycle == AppLifecycleState.detached) {
_heartbeatTimer?.cancel();
_heartbeatTimer = null;
return _setPresence('away');
}
return _fetch();
}
void _startHeartbeat() {
_heartbeatTimer?.cancel();
_heartbeatTimer = Timer.periodic(_heartbeatInterval, (_) {
_setPresence('online');
});
}
Future<String> _setPresence(String status) async {
final client = ref.read(relayClientProvider);
try {
await client.post('/api/presence', body: {'status': status});
return status;
} catch (_) {
return state.value ?? 'offline';
}
}
Future<String> _fetch() async {
final profile = ref.read(profileProvider).whenData((v) => v).value;
if (profile == null) return 'offline';
@@ -0,0 +1,86 @@
import 'dart:async';
import 'package:hooks_riverpod/hooks_riverpod.dart';
import '../../shared/relay/relay.dart';
import 'user_profile.dart';
/// In-memory cache of user profiles, fetched in batches from the relay.
class UserCacheNotifier extends Notifier<Map<String, UserProfile>> {
final Set<String> _pending = {};
Timer? _batchTimer;
@override
Map<String, UserProfile> build() {
ref.watch(relayClientProvider);
ref.onDispose(() {
_batchTimer?.cancel();
_batchTimer = null;
});
return {};
}
/// Request a profile for [pubkey]. Returns immediately from cache if
/// available, otherwise schedules a batch fetch.
UserProfile? get(String pubkey) {
final cached = state[pubkey];
if (cached != null) return cached;
_scheduleFetch(pubkey);
return null;
}
/// Preload profiles for a list of pubkeys (e.g. channel members).
void preload(List<String> pubkeys) {
final uncached = pubkeys
.map((pk) => pk.toLowerCase())
.where((pk) => !state.containsKey(pk) && !_pending.contains(pk))
.toList();
if (uncached.isEmpty) return;
_pending.addAll(uncached);
_batchTimer ??= Timer(const Duration(milliseconds: 50), _flushPending);
}
void _scheduleFetch(String pubkey) {
if (_pending.contains(pubkey)) return;
_pending.add(pubkey);
// Batch: wait 50ms to collect multiple lookups into one request.
_batchTimer ??= Timer(const Duration(milliseconds: 50), _flushPending);
}
Future<void> _flushPending() async {
_batchTimer = null;
if (_pending.isEmpty) return;
final pubkeys = _pending.toList();
_pending.clear();
try {
final client = ref.read(relayClientProvider);
final json =
await client.post('/api/users/batch', body: {'pubkeys': pubkeys})
as Map<String, dynamic>;
final profiles = json['profiles'] as Map<String, dynamic>? ?? {};
final updated = Map<String, UserProfile>.from(state);
for (final entry in profiles.entries) {
final data = entry.value as Map<String, dynamic>;
updated[entry.key.toLowerCase()] = UserProfile(
pubkey: entry.key.toLowerCase(),
displayName: data['display_name'] as String?,
avatarUrl: data['avatar_url'] as String?,
);
}
state = updated;
} catch (_) {
// Silently fail — we'll just show pubkeys.
}
}
}
final userCacheProvider =
NotifierProvider<UserCacheNotifier, Map<String, UserProfile>>(
UserCacheNotifier.new,
);
@@ -132,7 +132,7 @@ class SettingsPage extends HookConsumerWidget {
ref.read(authProvider.notifier).signOut();
},
style: FilledButton.styleFrom(
backgroundColor: context.colors.error,
backgroundColor: Theme.of(ctx).colorScheme.error,
),
child: const Text('Sign Out'),
),
+2 -7
View File
@@ -1,4 +1,3 @@
import 'package:flutter/foundation.dart';
import 'package:hooks_riverpod/hooks_riverpod.dart';
import '../relay/relay.dart';
@@ -16,12 +15,6 @@ class AuthState {
class AuthNotifier extends AsyncNotifier<AuthState> {
@override
Future<AuthState> build() async {
// If a token is provided via dart-define in debug mode, treat as
// pre-authenticated.
if (kDebugMode && Env.apiToken.isNotEmpty) {
return const AuthState(status: AuthStatus.authenticated);
}
final storage = CredentialStorage();
final creds = await storage.load();
if (creds == null) {
@@ -39,6 +32,7 @@ class AuthNotifier extends AsyncNotifier<AuthState> {
baseUrl: creds.relayUrl,
apiToken: creds.token,
devPubkey: null,
nsec: creds.nsec,
);
return AuthState(status: AuthStatus.authenticated, credentials: creds);
} on RelayException {
@@ -69,6 +63,7 @@ class AuthNotifier extends AsyncNotifier<AuthState> {
baseUrl: creds.relayUrl,
apiToken: creds.token,
devPubkey: null,
nsec: creds.nsec,
);
state = AsyncData(
+14 -1
View File
@@ -4,11 +4,13 @@ class StoredCredentials {
final String relayUrl;
final String token;
final String? pubkey;
final String? nsec;
const StoredCredentials({
required this.relayUrl,
required this.token,
this.pubkey,
this.nsec,
});
}
@@ -16,6 +18,7 @@ class CredentialStorage {
static const _keyRelayUrl = 'sprout_relay_url';
static const _keyToken = 'sprout_token';
static const _keyPubkey = 'sprout_pubkey';
static const _keyNsec = 'sprout_nsec';
final FlutterSecureStorage _storage;
@@ -28,7 +31,13 @@ class CredentialStorage {
if (relayUrl == null || token == null) return null;
final pubkey = await _storage.read(key: _keyPubkey);
return StoredCredentials(relayUrl: relayUrl, token: token, pubkey: pubkey);
final nsec = await _storage.read(key: _keyNsec);
return StoredCredentials(
relayUrl: relayUrl,
token: token,
pubkey: pubkey,
nsec: nsec,
);
}
Future<void> save(StoredCredentials credentials) async {
@@ -37,11 +46,15 @@ class CredentialStorage {
if (credentials.pubkey != null) {
await _storage.write(key: _keyPubkey, value: credentials.pubkey);
}
if (credentials.nsec != null) {
await _storage.write(key: _keyNsec, value: credentials.nsec);
}
}
Future<void> clear() async {
await _storage.delete(key: _keyRelayUrl);
await _storage.delete(key: _keyToken);
await _storage.delete(key: _keyPubkey);
await _storage.delete(key: _keyNsec);
}
}
@@ -0,0 +1,55 @@
import 'dart:async';
import 'package:connectivity_plus/connectivity_plus.dart';
import 'package:flutter/widgets.dart';
import 'package:hooks_riverpod/hooks_riverpod.dart';
import 'relay_session.dart';
/// Tracks the app lifecycle state and drives websocket connect/disconnect
/// behavior for mobile battery efficiency.
class AppLifecycleNotifier extends Notifier<AppLifecycleState> {
AppLifecycleListener? _listener;
StreamSubscription<List<ConnectivityResult>>? _connectivitySub;
@override
AppLifecycleState build() {
_listener = AppLifecycleListener(onStateChange: _onStateChange);
// Watch connectivity changes to trigger reconnect on network restore.
_connectivitySub = Connectivity().onConnectivityChanged.listen((results) {
final hasNetwork = results.any((r) => r != ConnectivityResult.none);
if (hasNetwork && state == AppLifecycleState.resumed) {
ref.read(relaySessionProvider.notifier).onAppResumed();
}
});
ref.onDispose(() {
_listener?.dispose();
_connectivitySub?.cancel();
});
return AppLifecycleState.resumed;
}
void _onStateChange(AppLifecycleState newState) {
state = newState;
final session = ref.read(relaySessionProvider.notifier);
switch (newState) {
case AppLifecycleState.resumed:
session.onAppResumed();
case AppLifecycleState.paused:
case AppLifecycleState.detached:
session.onAppPaused();
case AppLifecycleState.inactive:
case AppLifecycleState.hidden:
break; // Brief transition states — no action.
}
}
}
final appLifecycleProvider =
NotifierProvider<AppLifecycleNotifier, AppLifecycleState>(
AppLifecycleNotifier.new,
);
+138
View File
@@ -0,0 +1,138 @@
import 'package:flutter/foundation.dart';
/// Nostr event kind constants.
///
/// Keep in sync with `desktop/src/shared/constants/kinds.ts`.
abstract final class EventKind {
static const deletion = 5;
static const reaction = 7;
static const streamMessage = 9;
static const typingIndicator = 20002;
static const auth = 22242;
static const streamMessageV2 = 40002;
static const streamMessageEdit = 40003;
static const streamMessageDiff = 40008;
static const systemMessage = 40099;
static const forumPost = 45001;
static const forumComment = 45003;
/// Event kinds that represent channel activity (messages, edits, reactions,
/// deletions, system events). Matches the desktop's `CHANNEL_EVENT_KINDS`.
static const channelEventKinds = [
deletion, // 5
reaction, // 7
streamMessage, // 9
40001, // legacy pre-migration stream messages
streamMessageEdit, // 40003
streamMessageDiff, // 40008
systemMessage, // 40099
];
}
/// A Nostr event as defined by NIP-01.
@immutable
class NostrEvent {
final String id;
final String pubkey;
final int createdAt;
final int kind;
final List<List<String>> tags;
final String content;
final String sig;
const NostrEvent({
required this.id,
required this.pubkey,
required this.createdAt,
required this.kind,
required this.tags,
required this.content,
required this.sig,
});
factory NostrEvent.fromJson(Map<String, dynamic> json) {
return NostrEvent(
id: json['id'] as String,
pubkey: json['pubkey'] as String,
createdAt: json['created_at'] as int,
kind: json['kind'] as int,
tags: (json['tags'] as List<dynamic>)
.map((t) => (t as List<dynamic>).map((e) => e as String).toList())
.toList(),
content: json['content'] as String,
sig: json['sig'] as String,
);
}
Map<String, dynamic> toJson() => {
'id': id,
'pubkey': pubkey,
'created_at': createdAt,
'kind': kind,
'tags': tags,
'content': content,
'sig': sig,
};
/// Get the first value for a given tag key.
String? getTagValue(String key) {
for (final tag in tags) {
if (tag.isNotEmpty && tag[0] == key && tag.length > 1) {
return tag[1];
}
}
return null;
}
/// The channel/group ID from the `h` tag (NIP-29).
String? get channelId => getTagValue('h');
/// The parent event ID from the `e` tag.
String? get parentEventId => getTagValue('e');
@override
bool operator ==(Object other) =>
identical(this, other) || other is NostrEvent && id == other.id;
@override
int get hashCode => id.hashCode;
}
/// A NIP-01 subscription filter.
@immutable
class NostrFilter {
final List<int> kinds;
final int limit;
final int? since;
final int? until;
/// Tag filters, e.g. `{'#h': ['channel-id']}`.
final Map<String, List<String>> tags;
const NostrFilter({
required this.kinds,
this.limit = 100,
this.since,
this.until,
this.tags = const {},
});
/// Return a copy with an updated `since` value.
NostrFilter copyWithSince(int since) => NostrFilter(
kinds: kinds,
limit: limit,
since: since,
until: until,
tags: tags,
);
Map<String, dynamic> toJson() {
final json = <String, dynamic>{'kinds': kinds, 'limit': limit};
if (since != null) json['since'] = since;
if (until != null) json['until'] = until;
for (final entry in tags.entries) {
json[entry.key] = entry.value;
}
return json;
}
}
+4
View File
@@ -1,2 +1,6 @@
export 'app_lifecycle_provider.dart';
export 'nostr_models.dart';
export 'relay_client.dart';
export 'relay_provider.dart';
export 'relay_session.dart';
export 'relay_socket.dart';
+13
View File
@@ -63,6 +63,19 @@ class RelayClient {
return jsonDecode(response.body);
}
/// POST [path] with a pre-encoded string body. Returns the raw response body.
Future<String> postRaw(String path, {required String body}) async {
final response = await _http.post(
_uri(path),
headers: _headers,
body: body,
);
if (response.statusCode < 200 || response.statusCode >= 300) {
throw RelayException(response.statusCode, response.body);
}
return response.body;
}
void dispose() => _http.close();
}
+18 -1
View File
@@ -11,7 +11,22 @@ class RelayConfig {
/// Used when the relay has `SPROUT_REQUIRE_AUTH_TOKEN=false`.
final String? devPubkey;
const RelayConfig({required this.baseUrl, this.apiToken, this.devPubkey});
/// Nostr secret key (bech32 nsec) for signing NIP-42 AUTH events.
final String? nsec;
const RelayConfig({
required this.baseUrl,
this.apiToken,
this.devPubkey,
this.nsec,
});
/// Derive the websocket URL from the HTTP base URL.
String get wsUrl {
final uri = Uri.parse(baseUrl);
final scheme = uri.scheme == 'https' ? 'wss' : 'ws';
return uri.replace(scheme: scheme).toString();
}
}
/// Compile-time environment config via --dart-define.
@@ -44,11 +59,13 @@ class RelayConfigNotifier extends Notifier<RelayConfig> {
required String baseUrl,
required String? apiToken,
required String? devPubkey,
String? nsec,
}) {
state = RelayConfig(
baseUrl: baseUrl,
apiToken: apiToken,
devPubkey: devPubkey,
nsec: nsec,
);
}
}
+496
View File
@@ -0,0 +1,496 @@
import 'dart:async';
import 'dart:math';
import 'package:flutter/widgets.dart';
import 'package:hooks_riverpod/hooks_riverpod.dart';
import '../auth/auth.dart';
import 'nostr_models.dart';
import 'relay_provider.dart';
import 'relay_socket.dart';
// ---------------------------------------------------------------------------
// Session state
// ---------------------------------------------------------------------------
enum SessionStatus { disconnected, connecting, connected, reconnecting }
@immutable
class SessionState {
final SessionStatus status;
final int reconnectAttempt;
const SessionState({required this.status, this.reconnectAttempt = 0});
}
// ---------------------------------------------------------------------------
// Internal subscription types
// ---------------------------------------------------------------------------
class _HistorySubscription {
final List<NostrEvent> events = [];
final Completer<List<NostrEvent>> completer;
final Timer timeout;
_HistorySubscription({required this.completer, required this.timeout});
}
class _LiveSubscription {
final NostrFilter filter;
final void Function(NostrEvent) onEvent;
Completer<void>? readyCompleter;
int? lastSeenCreatedAt;
_LiveSubscription({
required this.filter,
required this.onEvent,
this.readyCompleter,
});
}
class _PendingEvent {
final Completer<NostrEvent> completer;
final Timer timeout;
_PendingEvent({required this.completer, required this.timeout});
}
class _BufferedEvent {
final String subId;
final NostrEvent event;
_BufferedEvent(this.subId, this.event);
}
// ---------------------------------------------------------------------------
// RelaySession — the core session manager
// ---------------------------------------------------------------------------
/// Manages websocket subscriptions, event batching, reconnection with replay,
/// and pending event tracking. Equivalent to the desktop's RelayClientSession.
class RelaySessionNotifier extends Notifier<SessionState> {
static const _baseReconnectDelayMs = 1000;
static const _maxReconnectDelayMs = 30000;
static const _eventBatchMs = 16;
static const _reconnectReplaySkewSeconds = 5;
RelaySocket? _socket;
final Map<String, _HistorySubscription> _historySubscriptions = {};
final Map<String, _LiveSubscription> _liveSubscriptions = {};
final Map<String, _PendingEvent> _pendingEvents = {};
final List<_BufferedEvent> _eventBuffer = [];
final Set<String> _recentEventIds = {};
Timer? _reconnectTimer;
Timer? _flushTimer;
Timer? _backgroundGraceTimer;
int _reconnectDelayMs = _baseReconnectDelayMs;
int _subIdCounter = 0;
bool _disposed = false;
@override
SessionState build() {
final config = ref.watch(relayConfigProvider);
final authState = ref.watch(authProvider);
ref.onDispose(_dispose);
// Auto-connect when authenticated and we have credentials.
final isAuthenticated = authState.value?.status == AuthStatus.authenticated;
if (isAuthenticated && config.apiToken != null) {
// Schedule connection after build completes.
Future.microtask(() => _connect(config));
}
return const SessionState(status: SessionStatus.disconnected);
}
// -------------------------------------------------------------------------
// Public API
// -------------------------------------------------------------------------
/// Fetch historical events matching [filter]. Sends REQ, collects events
/// until EOSE, then resolves. One-shot subscription.
Future<List<NostrEvent>> fetchHistory(
NostrFilter filter, {
Duration timeout = const Duration(seconds: 8),
}) {
final subId = _nextSubId('h');
final completer = Completer<List<NostrEvent>>();
final timer = Timer(timeout, () {
final sub = _historySubscriptions.remove(subId);
if (sub != null && !sub.completer.isCompleted) {
// Resolve with whatever we collected so far rather than failing.
sub.completer.complete(sub.events);
}
_sendClose(subId);
});
_historySubscriptions[subId] = _HistorySubscription(
completer: completer,
timeout: timer,
);
_sendReq(subId, filter);
return completer.future;
}
/// Subscribe to live events matching [filter]. Returns an unsubscribe
/// function. Live subscriptions survive reconnects — they are replayed with
/// `since: lastSeenCreatedAt - 5s` on reconnect.
Future<void Function()> subscribe(
NostrFilter filter,
void Function(NostrEvent) onEvent,
) async {
final subId = _nextSubId('l');
final readyCompleter = Completer<void>();
_liveSubscriptions[subId] = _LiveSubscription(
filter: filter,
onEvent: onEvent,
readyCompleter: readyCompleter,
);
_sendReq(subId, filter);
// Wait for EOSE or a short fallback timeout.
await readyCompleter.future.timeout(
const Duration(milliseconds: 500),
onTimeout: () {},
);
return () => _unsubscribe(subId);
}
/// Publish an event and wait for the relay's OK confirmation.
Future<NostrEvent> publish(
NostrEvent event, {
Duration timeout = const Duration(seconds: 8),
}) {
final completer = Completer<NostrEvent>();
final timer = Timer(timeout, () {
final pending = _pendingEvents.remove(event.id);
if (pending != null && !pending.completer.isCompleted) {
pending.completer.completeError(
TimeoutException(
'Event ${event.id} not acknowledged within $timeout',
),
);
}
});
_pendingEvents[event.id] = _PendingEvent(
completer: completer,
timeout: timer,
);
_socket?.send(['EVENT', event.toJson()]);
return completer.future;
}
/// Force a reconnect (e.g., returning from background).
Future<void> reconnect() async {
await _socket?.disconnect();
_reconnectDelayMs = _baseReconnectDelayMs;
final config = ref.read(relayConfigProvider);
await _connect(config);
}
/// Called by the app lifecycle provider when the app goes to background.
void onAppPaused() {
_backgroundGraceTimer?.cancel();
_backgroundGraceTimer = Timer(const Duration(seconds: 30), () {
_socket?.disconnect();
state = const SessionState(status: SessionStatus.disconnected);
});
}
/// Called by the app lifecycle provider when the app returns to foreground.
void onAppResumed() {
_backgroundGraceTimer?.cancel();
_backgroundGraceTimer = null;
if (state.status == SessionStatus.disconnected) {
_reconnectDelayMs = _baseReconnectDelayMs;
final config = ref.read(relayConfigProvider);
_connect(config);
}
}
// -------------------------------------------------------------------------
// Connection management
// -------------------------------------------------------------------------
Future<void> _connect(RelayConfig config) async {
if (_disposed) return;
if (_socket?.state == SocketState.connecting ||
_socket?.state == SocketState.authenticating) {
return;
}
state = SessionState(
status: SessionStatus.connecting,
reconnectAttempt: state.reconnectAttempt,
);
_socket?.dispose();
_socket = RelaySocket(
wsUrl: config.wsUrl,
nsec: config.nsec,
apiToken: config.apiToken,
onMessage: _handleMessage,
onConnected: _handleConnected,
onDisconnected: _handleDisconnected,
);
await _socket!.connect();
}
void _handleConnected() {
if (_disposed) return;
_reconnectDelayMs = _baseReconnectDelayMs;
state = const SessionState(status: SessionStatus.connected);
_replayLiveSubscriptions();
}
void _handleDisconnected(Object? error) {
if (_disposed) return;
_cancelAllHistory(error);
_rejectAllPending(error);
_eventBuffer.clear();
_flushTimer?.cancel();
_flushTimer = null;
_scheduleReconnect();
}
void _scheduleReconnect() {
if (_disposed) return;
if (_liveSubscriptions.isEmpty) {
state = const SessionState(status: SessionStatus.disconnected);
return;
}
final attempt = state.reconnectAttempt + 1;
state = SessionState(
status: SessionStatus.reconnecting,
reconnectAttempt: attempt,
);
_reconnectTimer?.cancel();
_reconnectTimer = Timer(Duration(milliseconds: _reconnectDelayMs), () {
_reconnectDelayMs = min(_reconnectDelayMs * 2, _maxReconnectDelayMs);
final config = ref.read(relayConfigProvider);
_connect(config);
});
}
/// Replay all live subscriptions after a reconnect, with a time skew to
/// catch events that occurred during the disconnect.
void _replayLiveSubscriptions() {
for (final entry in _liveSubscriptions.entries) {
final sub = entry.value;
final since = sub.lastSeenCreatedAt != null
? sub.lastSeenCreatedAt! - _reconnectReplaySkewSeconds
: null;
final filter = since != null
? sub.filter.copyWithSince(since)
: sub.filter;
_sendReq(entry.key, filter);
}
}
// -------------------------------------------------------------------------
// Message dispatch
// -------------------------------------------------------------------------
void _handleMessage(List<dynamic> data) {
if (data.isEmpty) return;
final type = data[0] as String;
switch (type) {
case 'EVENT':
_handleEvent(data);
case 'EOSE':
_handleEose(data);
case 'OK':
_handleOk(data);
}
}
void _handleEvent(List<dynamic> data) {
if (data.length < 3) return;
final subId = data[1] as String;
final eventJson = data[2] as Map<String, dynamic>;
final event = NostrEvent.fromJson(eventJson);
// History subscriptions accumulate immediately.
final historySub = _historySubscriptions[subId];
if (historySub != null) {
historySub.events.add(event);
return;
}
// Live subscriptions get batched.
final liveSub = _liveSubscriptions[subId];
if (liveSub != null) {
// Track last seen timestamp for reconnect replay.
if (liveSub.lastSeenCreatedAt == null ||
event.createdAt > liveSub.lastSeenCreatedAt!) {
liveSub.lastSeenCreatedAt = event.createdAt;
}
_eventBuffer.add(_BufferedEvent(subId, event));
_scheduleFlush();
}
}
void _handleEose(List<dynamic> data) {
if (data.length < 2) return;
final subId = data[1] as String;
// History subscription: resolve with collected events.
final historySub = _historySubscriptions.remove(subId);
if (historySub != null) {
historySub.timeout.cancel();
if (!historySub.completer.isCompleted) {
historySub.completer.complete(historySub.events);
}
_sendClose(subId);
return;
}
// Live subscription: signal ready.
final liveSub = _liveSubscriptions[subId];
if (liveSub != null &&
liveSub.readyCompleter != null &&
!liveSub.readyCompleter!.isCompleted) {
liveSub.readyCompleter!.complete();
liveSub.readyCompleter = null;
}
}
void _handleOk(List<dynamic> data) {
if (data.length < 3) return;
final eventId = data[1] as String;
final accepted = data[2] as bool;
final pending = _pendingEvents.remove(eventId);
if (pending == null) return;
pending.timeout.cancel();
if (accepted) {
// We don't have the full event here; create a minimal placeholder.
// The caller typically already has the event they published.
if (!pending.completer.isCompleted) {
pending.completer.complete(
NostrEvent(
id: eventId,
pubkey: '',
createdAt: 0,
kind: 0,
tags: [],
content: '',
sig: '',
),
);
}
} else {
final message = data.length > 3 ? data[3] as String : 'Event rejected';
if (!pending.completer.isCompleted) {
pending.completer.completeError(Exception(message));
}
}
}
// -------------------------------------------------------------------------
// Event batching (16ms flush interval)
// -------------------------------------------------------------------------
void _scheduleFlush() {
_flushTimer ??= Timer(
const Duration(milliseconds: _eventBatchMs),
_flushEventBuffer,
);
}
void _flushEventBuffer() {
_flushTimer = null;
if (_eventBuffer.isEmpty) return;
final batch = List<_BufferedEvent>.from(_eventBuffer);
_eventBuffer.clear();
for (final buffered in batch) {
// Deduplication: skip events we've already seen recently.
if (_recentEventIds.contains(buffered.event.id)) continue;
_recentEventIds.add(buffered.event.id);
// Cap the dedup set to prevent unbounded memory growth.
if (_recentEventIds.length > 5000) {
_recentEventIds.clear();
}
final sub = _liveSubscriptions[buffered.subId];
if (sub != null) {
sub.onEvent(buffered.event);
}
}
}
// -------------------------------------------------------------------------
// Helpers
// -------------------------------------------------------------------------
String _nextSubId(String prefix) {
_subIdCounter++;
return '$prefix-$_subIdCounter';
}
void _sendReq(String subId, NostrFilter filter) {
_socket?.send(['REQ', subId, filter.toJson()]);
}
void _sendClose(String subId) {
_socket?.send(['CLOSE', subId]);
}
void _unsubscribe(String subId) {
_liveSubscriptions.remove(subId);
_sendClose(subId);
}
void _cancelAllHistory(Object? error) {
for (final entry in _historySubscriptions.values) {
entry.timeout.cancel();
if (!entry.completer.isCompleted) {
entry.completer.completeError(error ?? Exception('Connection lost'));
}
}
_historySubscriptions.clear();
}
void _rejectAllPending(Object? error) {
for (final entry in _pendingEvents.values) {
entry.timeout.cancel();
if (!entry.completer.isCompleted) {
entry.completer.completeError(error ?? Exception('Connection lost'));
}
}
_pendingEvents.clear();
}
void _dispose() {
_disposed = true;
_reconnectTimer?.cancel();
_flushTimer?.cancel();
_backgroundGraceTimer?.cancel();
_cancelAllHistory(null);
_rejectAllPending(null);
_socket?.dispose();
_socket = null;
}
}
final relaySessionProvider =
NotifierProvider<RelaySessionNotifier, SessionState>(
RelaySessionNotifier.new,
);
+230
View File
@@ -0,0 +1,230 @@
import 'dart:async';
import 'dart:convert';
import 'package:nostr/nostr.dart' as nostr;
import 'package:web_socket_channel/web_socket_channel.dart';
import 'nostr_models.dart';
/// Low-level websocket connection with NIP-42 authentication.
///
/// Handles the raw websocket lifecycle: connect, authenticate via NIP-42
/// challenge/response, send/receive JSON frames, and disconnect.
///
/// Does NOT handle reconnection — that is [RelaySessionNotifier]'s job.
enum SocketState { disconnected, connecting, authenticating, connected }
class RelaySocket {
final String _wsUrl;
final String? _nsec;
final String? _apiToken;
final void Function(List<dynamic> message) _onMessage;
final void Function() _onConnected;
final void Function(Object? error) _onDisconnected;
WebSocketChannel? _channel;
StreamSubscription<dynamic>? _subscription;
SocketState _state = SocketState.disconnected;
Completer<void>? _authCompleter;
Timer? _authTimeout;
String? _pendingAuthEventId;
SocketState get state => _state;
RelaySocket({
required String wsUrl,
required String? nsec,
required String? apiToken,
required void Function(List<dynamic> message) onMessage,
required void Function() onConnected,
required void Function(Object? error) onDisconnected,
}) : _wsUrl = wsUrl,
_nsec = nsec,
_apiToken = apiToken,
_onMessage = onMessage,
_onConnected = onConnected,
_onDisconnected = onDisconnected;
/// Connect to the relay and complete NIP-42 authentication.
Future<void> connect() async {
if (_state != SocketState.disconnected) return;
_state = SocketState.connecting;
try {
_channel = WebSocketChannel.connect(Uri.parse(_wsUrl));
await _channel!.ready;
} catch (e) {
_state = SocketState.disconnected;
_onDisconnected(e);
return;
}
_state = SocketState.authenticating;
_authCompleter = Completer<void>();
_subscription = _channel!.stream.listen(
_handleRawMessage,
onError: (Object error) {
_failAuth(error);
_resetConnection();
_onDisconnected(error);
},
onDone: () {
_failAuth(null);
_resetConnection();
_onDisconnected(null);
},
);
// Wait for auth to complete (or timeout).
_authTimeout = Timer(const Duration(seconds: 8), () {
if (_authCompleter != null && !_authCompleter!.isCompleted) {
_authCompleter!.completeError(
TimeoutException('NIP-42 auth timed out after 8s'),
);
}
});
try {
await _authCompleter!.future;
_authTimeout?.cancel();
_state = SocketState.connected;
_onConnected();
} catch (e) {
_authTimeout?.cancel();
await disconnect();
_onDisconnected(e);
}
}
/// Send a raw JSON array over the websocket.
void send(List<dynamic> payload) {
_channel?.sink.add(jsonEncode(payload));
}
/// Gracefully close the connection.
Future<void> disconnect() async {
_resetConnection();
final channel = _channel;
_channel = null;
if (channel != null) {
await channel.sink.close();
}
}
void dispose() {
_resetConnection();
_channel?.sink.close();
_channel = null;
}
void _resetConnection() {
_state = SocketState.disconnected;
_subscription?.cancel();
_subscription = null;
_authTimeout?.cancel();
_authTimeout = null;
_pendingAuthEventId = null;
}
void _failAuth(Object? error) {
if (_authCompleter != null && !_authCompleter!.isCompleted) {
_authCompleter!.completeError(error ?? Exception('Connection closed'));
}
}
void _handleRawMessage(dynamic raw) {
final String text;
if (raw is String) {
text = raw;
} else {
return; // Binary frames are not part of the Nostr protocol.
}
final List<dynamic> data;
try {
data = jsonDecode(text) as List<dynamic>;
} catch (_) {
return; // Malformed JSON.
}
if (data.isEmpty) return;
final type = data[0] as String;
switch (type) {
case 'AUTH':
_handleAuthChallenge(data);
case 'OK':
_handleOk(data);
default:
// Pass EVENT, EOSE, NOTICE, etc. upstream.
_onMessage(data);
}
}
/// Handle the relay's AUTH challenge: sign a kind:22242 event and respond.
void _handleAuthChallenge(List<dynamic> data) {
if (data.length < 2) return;
final challenge = data[1] as String;
if (_nsec == null) {
_failAuth(Exception('No nsec available for NIP-42 auth'));
return;
}
try {
// Decode bech32 nsec to hex private key.
final privkeyHex = nostr.Nip19.decodePrivkey(_nsec);
if (privkeyHex.isEmpty) {
_failAuth(Exception('Invalid nsec'));
return;
}
// Build the auth tags.
final tags = <List<String>>[
['relay', _wsUrl],
['challenge', challenge],
if (_apiToken != null) ['auth_token', _apiToken],
];
// Create and sign the kind:22242 AUTH event.
final event = nostr.Event.from(
kind: EventKind.auth,
content: '',
tags: tags,
privkey: privkeyHex,
);
_pendingAuthEventId = event.id;
send(['AUTH', event.toJson()]);
} catch (e) {
_failAuth(e);
}
}
/// Handle OK frames. During auth, complete the auth flow.
void _handleOk(List<dynamic> data) {
if (data.length < 3) return;
final eventId = data[1] as String;
final accepted = data[2] as bool;
// Check if this OK is for our pending AUTH event.
if (_pendingAuthEventId != null && eventId == _pendingAuthEventId) {
_pendingAuthEventId = null;
if (accepted) {
if (_authCompleter != null && !_authCompleter!.isCompleted) {
_authCompleter!.complete();
}
} else {
final message = data.length > 3
? data[3] as String
: 'Auth rejected by relay';
_failAuth(Exception(message));
}
return;
}
// Pass non-auth OK frames upstream for pending event tracking.
_onMessage(data);
}
}
+81 -1
View File
@@ -57,6 +57,22 @@ packages:
url: "https://pub.dev"
source: hosted
version: "2.13.1"
bech32:
dependency: transitive
description:
name: bech32
sha256: "156cbace936f7720c79a79d16a03efad343b1ef17106716e04b8b8e39f99f7f7"
url: "https://pub.dev"
source: hosted
version: "0.2.2"
bip340:
dependency: transitive
description:
name: bip340
sha256: "2a92f6ed68959f75d67c9a304c17928b9c9449587d4f75ee68f34152f7f69e87"
url: "https://pub.dev"
source: hosted
version: "0.2.0"
boolean_selector:
dependency: transitive
description:
@@ -137,6 +153,22 @@ packages:
url: "https://pub.dev"
source: hosted
version: "1.19.1"
connectivity_plus:
dependency: "direct main"
description:
name: connectivity_plus
sha256: b5e72753cf63becce2c61fd04dfe0f1c430cc5278b53a1342dc5ad839eab29ec
url: "https://pub.dev"
source: hosted
version: "6.1.5"
connectivity_plus_platform_interface:
dependency: transitive
description:
name: connectivity_plus_platform_interface
sha256: "3c09627c536d22fd24691a905cdd8b14520de69da52c7a97499c8be5284a32ed"
url: "https://pub.dev"
source: hosted
version: "2.1.0"
convert:
dependency: transitive
description:
@@ -193,6 +225,14 @@ packages:
url: "https://pub.dev"
source: hosted
version: "3.1.3"
dbus:
dependency: transitive
description:
name: dbus
sha256: d0c98dcd4f5169878b6cf8f6e0a52403a9dff371a3e2f019697accbf6f44a270
url: "https://pub.dev"
source: hosted
version: "0.7.12"
fake_async:
dependency: transitive
description:
@@ -520,6 +560,14 @@ packages:
url: "https://pub.dev"
source: hosted
version: "0.17.6"
nm:
dependency: transitive
description:
name: nm
sha256: "2c9aae4127bdc8993206464fcc063611e0e36e72018696cd9631023a31b24254"
url: "https://pub.dev"
source: hosted
version: "0.5.0"
node_preamble:
dependency: transitive
description:
@@ -528,6 +576,14 @@ packages:
url: "https://pub.dev"
source: hosted
version: "2.0.2"
nostr:
dependency: "direct main"
description:
name: nostr
sha256: a99942e4eedd5823d16f42e6df96240488028666a329ee1047552f79db564123
url: "https://pub.dev"
source: hosted
version: "1.5.0"
objective_c:
dependency: transitive
description:
@@ -600,6 +656,14 @@ packages:
url: "https://pub.dev"
source: hosted
version: "2.3.0"
petitparser:
dependency: transitive
description:
name: petitparser
sha256: "91bd59303e9f769f108f8df05e371341b15d59e995e6806aefab827b58336675"
url: "https://pub.dev"
source: hosted
version: "7.0.2"
platform:
dependency: transitive
description:
@@ -616,6 +680,14 @@ packages:
url: "https://pub.dev"
source: hosted
version: "2.1.8"
pointycastle:
dependency: transitive
description:
name: pointycastle
sha256: "4be0097fcf3fd3e8449e53730c631200ebc7b88016acecab2b0da2f0149222fe"
url: "https://pub.dev"
source: hosted
version: "3.9.1"
pool:
dependency: transitive
description:
@@ -918,7 +990,7 @@ packages:
source: hosted
version: "1.0.1"
web_socket_channel:
dependency: transitive
dependency: "direct main"
description:
name: web_socket_channel
sha256: d645757fb0f4773d602444000a8131ff5d48c9e47adfe9772652dd1a4f2d45c8
@@ -949,6 +1021,14 @@ packages:
url: "https://pub.dev"
source: hosted
version: "1.1.0"
xml:
dependency: transitive
description:
name: xml
sha256: "971043b3a0d3da28727e40ed3e0b5d18b742fa5a68665cca88e74b7876d5e025"
url: "https://pub.dev"
source: hosted
version: "6.6.1"
yaml:
dependency: transitive
description:
+3
View File
@@ -16,6 +16,9 @@ dependencies:
flutter_secure_storage: ^9.2.4
mobile_scanner: ^6.0.2
shared_preferences: ^2.5.5
web_socket_channel: ^3.0.1
connectivity_plus: ^6.1.4
nostr: ^1.5.0
dev_dependencies:
flutter_test: