ensure minimum threshold

lib/src/database/daos/user_discovery.dao.dart

@@ -10,6 +10,15 @@ part 'user_discovery.dao.g.dart';
 typedef AnnouncedUsersWithRelations =
     Map<UserDiscoveryAnnouncedUser, List<(Contact, DateTime?)>>;
 
+const _minimumUserDiscoveryThreshold = 3;
+
+int _currentUserDiscoveryThreshold() {
+  final configured = userService.currentUser.userDiscoveryThreshold;
+  return configured < _minimumUserDiscoveryThreshold
+      ? _minimumUserDiscoveryThreshold
+      : configured;
+}
+
 @DriftAccessor(
   tables: [
     UserDiscoveryAnnouncedUsers,
@@ -53,8 +62,10 @@ class UserDiscoveryDao extends DatabaseAccessor<TwonlyDB>
         .map((row) => row.readTable(userDiscoveryAnnouncedUsers))
         .toList();
   }
+
   Stream<AnnouncedUsersWithRelations> watchAllAnnouncedUsersWithRelations() {
-    final query = select(userDiscoveryAnnouncedUsers).join([
+    final query =
+        select(userDiscoveryAnnouncedUsers).join([
           innerJoin(
             userDiscoveryUserRelations,
             userDiscoveryUserRelations.announcedUserId.equalsExp(
@@ -67,7 +78,13 @@ Stream<AnnouncedUsersWithRelations> watchAllAnnouncedUsersWithRelations() {
               userDiscoveryUserRelations.fromContactId,
             ),
           ),
-    ])..where(userDiscoveryAnnouncedUsers.username.isNotNull());
+        ])..where(
+          userDiscoveryAnnouncedUsers.username.isNotNull() &
+              contacts.accepted.equals(true) &
+              contacts.blocked.equals(false) &
+              contacts.deletedByUser.equals(false) &
+              contacts.accountDeleted.equals(false),
+        );
 
     return query.watch().map((rows) {
       // ignore: omit_local_variable_types
@@ -89,6 +106,9 @@ Stream<AnnouncedUsersWithRelations> watchAllAnnouncedUsersWithRelations() {
         results[user]!.add(relationData);
       }
 
+      final threshold = _currentUserDiscoveryThreshold();
+      results.removeWhere((user, relations) => relations.length < threshold);
+
       Log.info('results = ${results.length}');
 
       return results;
@@ -120,6 +140,10 @@ Stream<AnnouncedUsersWithRelations> watchAllAnnouncedUsersWithRelations() {
         ])..where(
           userDiscoveryAnnouncedUsers.username.isNotNull() &
               userDiscoveryAnnouncedUsers.isHidden.equals(false) &
+              contacts.accepted.equals(true) &
+              contacts.blocked.equals(false) &
+              contacts.deletedByUser.equals(false) &
+              contacts.accountDeleted.equals(false) &
               (announcedContact.userId.isNull() |
                   announcedContact.deletedByUser.equals(true)),
         );
@@ -144,7 +168,7 @@ Stream<AnnouncedUsersWithRelations> watchAllAnnouncedUsersWithRelations() {
         results[user]!.add(relationData);
       }
 
-      final threshold = userService.currentUser.userDiscoveryThreshold;
+      final threshold = _currentUserDiscoveryThreshold();
       results.removeWhere((user, relations) => relations.length < threshold);
 
       return results;
@@ -161,6 +185,12 @@ Stream<AnnouncedUsersWithRelations> watchAllAnnouncedUsersWithRelations() {
               userDiscoveryAnnouncedUsers.announcedUserId,
             ),
           ),
+          innerJoin(
+            contacts,
+            contacts.userId.equalsExp(
+              userDiscoveryUserRelations.fromContactId,
+            ),
+          ),
           leftOuterJoin(
             announcedContact,
             announcedContact.userId.equalsExp(
@@ -172,6 +202,10 @@ Stream<AnnouncedUsersWithRelations> watchAllAnnouncedUsersWithRelations() {
           userDiscoveryAnnouncedUsers.username.isNotNull() &
               userDiscoveryAnnouncedUsers.wasShownToTheUser.equals(false) &
               userDiscoveryAnnouncedUsers.isHidden.equals(false) &
+              contacts.accepted.equals(true) &
+              contacts.blocked.equals(false) &
+              contacts.deletedByUser.equals(false) &
+              contacts.accountDeleted.equals(false) &
               (announcedContact.userId.isNull() |
                   announcedContact.deletedByUser.equals(true)),
         );
@@ -186,7 +220,7 @@ Stream<AnnouncedUsersWithRelations> watchAllAnnouncedUsersWithRelations() {
             (relationCounts[announcedUserId] ?? 0) + 1;
       }
 
-      final threshold = userService.currentUser.userDiscoveryThreshold;
+      final threshold = _currentUserDiscoveryThreshold();
       return relationCounts.values.where((count) => count >= threshold).length;
     });
   }

rust/src/user_config.rs

@@ -12,6 +12,8 @@ use std::io::Write;
 use std::path::PathBuf;
 use std::sync::{OnceLock, RwLock};
 
+pub(crate) const MIN_USER_DISCOVERY_THRESHOLD: u8 = 3;
+
 mod defaults {
     pub fn true_value() -> bool {
         true
@@ -30,7 +32,7 @@ mod defaults {
     }
 
     pub fn user_discovery_threshold() -> u8 {
-        3
+        super::MIN_USER_DISCOVERY_THRESHOLD
     }
 
     pub fn passwordless_threshold() -> i64 {
@@ -355,6 +357,12 @@ pub struct UserConfig {
 }
 
 impl UserConfig {
+    fn normalize(&mut self) {
+        self.user_discovery_threshold = self
+            .user_discovery_threshold
+            .max(MIN_USER_DISCOVERY_THRESHOLD);
+    }
+
     fn path(context: &Context) -> PathBuf {
         PathBuf::from(&context.config.data_dir)
             .join("keyvalue")
@@ -378,12 +386,14 @@ impl UserConfig {
             return Ok(None);
         }
         let file = File::open(&path)?;
-        serde_json::from_reader(file).map(Some).map_err(|error| {
+        let mut config: Self = serde_json::from_reader(file).map_err(|error| {
             TwonlyError::Generic(format!(
                 "invalid user configuration {}: {error}",
                 path.display()
             ))
-        })
+        })?;
+        config.normalize();
+        Ok(Some(config))
     }
 
     /// Validates and atomically persists the complete user configuration.
@@ -405,6 +415,7 @@ impl UserConfig {
         let mut config = Self::load_from_unlocked(context)?
             .ok_or_else(|| twonly_error!("user configuration is unavailable"))?;
         mutate(&mut config);
+        config.normalize();
         Self::save_unlocked(context, &config)?;
         Ok(config)
     }
@@ -433,13 +444,16 @@ impl UserConfig {
     }
 
     fn save_json_unlocked(context: &Context, json: &str) -> Result<String> {
-        let config: Self = serde_json::from_str(json).map_err(|error| {
+        let mut config: Self = serde_json::from_str(json).map_err(|error| {
             TwonlyError::Generic(format!("invalid user configuration update: {error}"))
         })?;
+        config.normalize();
         Self::save_unlocked(context, &config)
     }
 
     fn save_unlocked(context: &Context, config: &Self) -> Result<String> {
+        let mut config = config.clone();
+        config.normalize();
         let normalized = serde_json::to_string(&config)?;
         let path = Self::path(context);
         let parent = path
@@ -457,7 +471,7 @@ impl UserConfig {
 
 #[cfg(test)]
 mod tests {
-    use super::{merge_changed_fields, UserConfig};
+    use super::{merge_changed_fields, UserConfig, MIN_USER_DISCOVERY_THRESHOLD};
 
     #[test]
     fn parses_only_rust_fields_and_applies_defaults() {
@@ -474,12 +488,36 @@ mod tests {
         assert_eq!(config.user_id, 42);
         assert_eq!(config.username, "alice");
         assert_eq!(config.required_send_images, 4);
+        assert_eq!(
+            config.user_discovery_threshold,
+            MIN_USER_DISCOVERY_THRESHOLD
+        );
 
         assert!(config.can_use_login_token_for_auth);
         assert!(!config.is_user_discovery_enabled);
         assert_eq!(config.last_server_message_at, None);
     }
 
+    #[test]
+    fn normalizes_legacy_user_discovery_threshold() {
+        let mut config: UserConfig = serde_json::from_str(
+            r#"{
+                "userId": 42,
+                "username": "alice",
+                "displayName": "Alice",
+                "userDiscoveryThreshold": 2
+            }"#,
+        )
+        .unwrap();
+
+        config.normalize();
+
+        assert_eq!(
+            config.user_discovery_threshold,
+            MIN_USER_DISCOVERY_THRESHOLD
+        );
+    }
+
     #[test]
     fn merges_only_fields_changed_by_a_stale_snapshot() {
         let mut current = serde_json::from_str(r#"{"theme":"dark","counter":2}"#).unwrap();

rust/src/user_discovery.rs

@@ -18,6 +18,7 @@ use tokio::sync::{Mutex, RwLock};
 use crate::database::signal::Database;
 use crate::keys::KeyManager;
 use crate::error::{Result, TwonlyError};
+use crate::user_config::MIN_USER_DISCOVERY_THRESHOLD;
 use crate::user_discovery::user_discovery_message::{UserDiscoveryAnnouncement, UserDiscoveryPromotion};
 use crate::user_discovery::user_discovery_message::user_discovery_promotion::AnnouncementShareDecrypted;
 use crate::user_discovery::user_discovery_message::user_discovery_promotion::announcement_share_decrypted::SignedData;
@@ -26,6 +27,10 @@ use crate::user_discovery::user_discovery_message::user_discovery_promotion::ann
 /// the types.proto
 pub type UserID = i64;
 
+fn is_supported_threshold(threshold: u32) -> bool {
+    (u32::from(MIN_USER_DISCOVERY_THRESHOLD)..=u32::from(u8::MAX)).contains(&threshold)
+}
+
 include!(concat!(env!("OUT_DIR"), "/user_discovery.rs"));
 
 #[derive(Clone, sqlx::FromRow)]
@@ -263,6 +268,11 @@ impl UserDiscovery {
         share_promotion: bool,
         t: &mut sqlx::Transaction<'_, sqlx::Sqlite>,
     ) -> Result<()> {
+        if threshold < MIN_USER_DISCOVERY_THRESHOLD {
+            return Err(TwonlyError::UserDiscoveryStore(format!(
+                "user discovery threshold must be at least {MIN_USER_DISCOVERY_THRESHOLD}"
+            )));
+        }
         tracing::info!("Protocols: initialize_or_update started, getting config from store");
         let config = match self.read_config() {
             Ok(config) => {
@@ -683,6 +693,16 @@ impl UserDiscovery {
     ) -> Result<()> {
         tracing::info!("Got a user discovery announcement from {contact_id}.");
 
+        if !is_supported_threshold(uda.threshold) {
+            tracing::warn!(
+                "Ignoring user-discovery announcement with unsupported threshold {} (expected {}..={}).",
+                uda.threshold,
+                MIN_USER_DISCOVERY_THRESHOLD,
+                u8::MAX
+            );
+            return Ok(());
+        }
+
         if uda.threshold as usize != uda.verification_shares.len() + 1 {
             tracing::warn!(
                 "UDA contains to few shares to verify: {} != {} + 1.",
@@ -835,6 +855,16 @@ impl UserDiscovery {
     ) -> Result<()> {
         tracing::debug!("Received a new UDP with public_id = {}.", &udp.public_id);
 
+        if !is_supported_threshold(udp.threshold) {
+            tracing::warn!(
+                "Ignoring user-discovery promotion with unsupported threshold {} (expected {}..={}).",
+                udp.threshold,
+                MIN_USER_DISCOVERY_THRESHOLD,
+                u8::MAX
+            );
+            return Ok(());
+        }
+
         if udp.announcement_share.is_empty() {
             tracing::info!("Got empty announcement share. Ignoring it..");
             return Ok(());
@@ -1101,7 +1131,7 @@ impl UserDiscovery {
 impl Default for UserDiscoveryConfig {
     fn default() -> Self {
         Self {
-            threshold: 2,
+            threshold: MIN_USER_DISCOVERY_THRESHOLD,
             total_number_of_shares: 255,
             announcement_version: 0,
             promotion_version: 0,
@@ -1112,3 +1142,20 @@ impl Default for UserDiscoveryConfig {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{is_supported_threshold, MIN_USER_DISCOVERY_THRESHOLD};
+
+    #[test]
+    fn only_accepts_supported_discovery_thresholds() {
+        assert!(!is_supported_threshold(u32::from(
+            MIN_USER_DISCOVERY_THRESHOLD - 1
+        )));
+        assert!(is_supported_threshold(u32::from(
+            MIN_USER_DISCOVERY_THRESHOLD
+        )));
+        assert!(is_supported_threshold(u32::from(u8::MAX)));
+        assert!(!is_supported_threshold(u32::from(u8::MAX) + 1));
+    }
+}

test/drift/user_discovery_dao_test.dart

@@ -0,0 +1,115 @@
+import 'package:drift/drift.dart';
+import 'package:drift/native.dart';
+import 'package:flutter_test/flutter_test.dart';
+import 'package:twonly/locator.dart';
+import 'package:twonly/src/database/twonly.db.dart';
+import 'package:twonly/src/services/user.service.dart';
+
+import '../mocks/user_config.dart';
+
+void main() {
+  TestWidgetsFlutterBinding.ensureInitialized();
+
+  setUp(() async {
+    await locator.reset();
+    locator
+      ..registerSingleton<TwonlyDB>(
+        TwonlyDB.forTesting(
+          DatabaseConnection(
+            NativeDatabase.memory(),
+            closeStreamsSynchronously: true,
+          ),
+        ),
+      )
+      ..registerSingleton<UserService>(UserService());
+
+    userService.currentUser = testUserConfig(
+      userId: 1,
+      username: 'observer',
+      displayName: 'Observer',
+      subscriptionPlan: 'Free',
+      currentSetupPage: null,
+      appVersion: 119,
+    );
+    userService.isUserCreated = true;
+
+    await twonlyDB.customStatement(
+      "INSERT INTO user_discovery_announced_users "
+      "(announced_user_id, announced_public_key, public_id, username) "
+      "VALUES (99, x'01', 999, 'suggested')",
+    );
+    for (final id in [10, 11, 12]) {
+      await twonlyDB.customStatement(
+        'INSERT INTO contacts(user_id, username, accepted) '
+        "VALUES ($id, 'friend$id', 1)",
+      );
+    }
+  });
+
+  tearDown(() async {
+    await twonlyDB.close();
+  });
+
+  Future<void> addRelation(int friendId) {
+    return twonlyDB.customStatement(
+      'INSERT INTO user_discovery_user_relations '
+      '(announced_user_id, from_contact_id) VALUES (99, $friendId)',
+    );
+  }
+
+  test('suggestions and request context require three valid friends', () async {
+    await addRelation(10);
+    await addRelation(11);
+
+    expect(
+      await twonlyDB.userDiscoveryDao
+          .watchNewAnnouncedUsersWithRelations()
+          .first,
+      isEmpty,
+    );
+    expect(
+      await twonlyDB.userDiscoveryDao
+          .watchAllAnnouncedUsersWithRelations()
+          .first,
+      isEmpty,
+    );
+    expect(
+      await twonlyDB.userDiscoveryDao
+          .watchNewAnnouncementsWithDataCount()
+          .first,
+      0,
+    );
+
+    await addRelation(12);
+
+    expect(
+      await twonlyDB.userDiscoveryDao
+          .watchNewAnnouncedUsersWithRelations()
+          .first,
+      hasLength(1),
+    );
+    expect(
+      await twonlyDB.userDiscoveryDao
+          .watchAllAnnouncedUsersWithRelations()
+          .first,
+      hasLength(1),
+    );
+    expect(
+      await twonlyDB.userDiscoveryDao
+          .watchNewAnnouncementsWithDataCount()
+          .first,
+      1,
+    );
+
+    await twonlyDB.customStatement(
+      'UPDATE contacts SET blocked = 1 WHERE user_id = 12',
+    );
+
+    expect(
+      await twonlyDB.userDiscoveryDao
+          .watchNewAnnouncedUsersWithRelations()
+          .first,
+      isEmpty,
+    );
+  });
+}