web: add ?since= parameter to api/messages (#512)

* web: add ?since= parameter to api/messages

* app: add ?since= parameter to api/messages

README.md

@@ -130,7 +130,7 @@ The web app contains an API:
 
 * GET `/api/nodes?limit=100` - returns the latest 100 nodes reported to the app
 * GET `/api/positions?limit=100` - returns the latest 100 position data
-* GET `/api/messages?limit=100&encrypted=false` - returns the latest 100 messages (disabled when `PRIVATE=1`)
+* GET `/api/messages?limit=100&encrypted=false&since=0` - returns the latest 100 messages newer than the provided unix timestamp (defaults to `since=0` to return full history; disabled when `PRIVATE=1`)
 * GET `/api/telemetry?limit=100` - returns the latest 100 telemetry data
 * GET `/api/neighbors?limit=100` - returns the latest 100 neighbor tuples
 * GET `/api/instances` - returns known potato-mesh instances in other locations

app/lib/main.dart

@@ -1144,8 +1144,7 @@ class MeshRepository implements MeshNodeResolver {
     }
   }
 
-  /// Fetches a complete messages list on first load, falling back to a smaller
-  /// refresh window on subsequent calls.
+  /// Fetches messages, using a full sync initially and incremental loads later.
   Future<List<MeshMessage>> loadMessages({required String domain}) async {
     await _ensureStore();
     final key = _domainKey(domain);
@@ -1485,9 +1484,10 @@ class MeshRepository implements MeshNodeResolver {
     final key = _domainKey(domain);
     final alreadyLoaded = _messagesLoaded[key] ?? false;
     final initialFetch = forceFull || !alreadyLoaded;
-    final limit = initialFetch ? 1000 : 100;
+    final since = initialFetch ? 0 : _latestRxTimeSeconds(domain);
+    const limit = 1000;
 
-    final uri = _buildMessagesUri(domain, limit: limit);
+    final uri = _buildMessagesUri(domain, since: since, limit: limit);
     _logHttp('GET $uri');
     final resp = await client.get(uri).timeout(_requestTimeout);
     _logHttp('HTTP ${resp.statusCode} $uri');
@@ -1543,6 +1543,33 @@ class MeshRepository implements MeshNodeResolver {
     return sorted;
   }
 
+  int _latestRxTimeSeconds(String domain) {
+    final key = _domainKey(domain);
+    final messages = _messagesByDomain[key];
+    if (messages == null || messages.isEmpty) return 0;
+
+    var maxSeconds = 0;
+    for (final message in messages) {
+      final rxMillis = message.rxTime?.toUtc().millisecondsSinceEpoch;
+      final candidate = message.rxTimeSeconds ??
+          (rxMillis != null ? rxMillis ~/ 1000 : null) ??
+          _coerceIsoSeconds(message.rxIso);
+      if (candidate != null && candidate > maxSeconds) {
+        maxSeconds = candidate;
+      }
+    }
+    return maxSeconds;
+  }
+
+  int? _coerceIsoSeconds(String rxIso) {
+    if (rxIso.isEmpty) return null;
+    try {
+      return DateTime.parse(rxIso).toUtc().millisecondsSinceEpoch ~/ 1000;
+    } catch (_) {
+      return null;
+    }
+  }
+
   Future<void> _hydrateMissingNodes({
     required String domain,
     required List<MeshMessage> messages,
@@ -2554,6 +2581,7 @@ class _SettingsScreenState extends State<SettingsScreen> {
 /// Representation of a single mesh message returned by the PotatoMesh API.
 class MeshMessage {
   final int id;
+  final int? rxTimeSeconds;
   final DateTime? rxTime;
   final String rxIso;
   final String fromId;
@@ -2570,6 +2598,7 @@ class MeshMessage {
   /// Creates a [MeshMessage] with all properties parsed from the API response.
   MeshMessage({
     required this.id,
+    this.rxTimeSeconds,
     required this.rxTime,
     required this.rxIso,
     required this.fromId,
@@ -2586,15 +2615,32 @@ class MeshMessage {
 
   /// Parses a [MeshMessage] from the raw JSON map returned by the API.
   factory MeshMessage.fromJson(Map<String, dynamic> json) {
+    int? parsedSeconds;
+    final rawSeconds = json['rx_time'];
+    if (rawSeconds != null) {
+      parsedSeconds = int.tryParse(rawSeconds.toString());
+    }
+
     DateTime? parsedTime;
     if (json['rx_iso'] is String) {
       try {
-        parsedTime = DateTime.parse(json['rx_iso'] as String).toLocal();
+        final parsedIso = DateTime.parse(json['rx_iso'] as String);
+        parsedTime = parsedIso.toLocal();
+        parsedSeconds ??= parsedIso.toUtc().millisecondsSinceEpoch ~/ 1000;
       } catch (_) {
         parsedTime = null;
       }
     }
 
+    final parsedMillis = parsedTime?.toUtc().millisecondsSinceEpoch;
+    if (parsedMillis != null && parsedSeconds == null) {
+      parsedSeconds = parsedMillis ~/ 1000;
+    }
+    parsedTime ??= parsedSeconds != null
+        ? DateTime.fromMillisecondsSinceEpoch(parsedSeconds * 1000, isUtc: true)
+            .toLocal()
+        : null;
+
     double? parseDouble(dynamic v) {
       if (v == null) return null;
       if (v is num) return v.toDouble();
@@ -2609,6 +2655,7 @@ class MeshMessage {
 
     return MeshMessage(
       id: parseInt(json['id']) ?? 0,
+      rxTimeSeconds: parsedSeconds,
       rxTime: parsedTime,
       rxIso: json['rx_iso']?.toString() ?? '',
       fromId: json['from_id']?.toString() ?? '',
@@ -2645,6 +2692,7 @@ class MeshMessage {
   Map<String, dynamic> toJson() {
     return {
       'id': id,
+      'rx_time': rxTimeSeconds,
       'rx_iso': rxIso,
       'from_id': fromId,
       'node_id': nodeId,
@@ -2811,11 +2859,12 @@ class MeshNode {
 }
 
 /// Build a messages API URI for a given domain or absolute URL.
-Uri _buildMessagesUri(String domain, {int limit = 1000}) {
+Uri _buildMessagesUri(String domain, {int since = 0, int limit = 1000}) {
   final trimmed = domain.trim();
   final params = {
     'limit': limit.toString(),
     'encrypted': 'false',
+    'since': since.toString(),
   };
   if (trimmed.isEmpty) {
     return Uri.https('potatomesh.net', '/api/messages', params);
@@ -2952,9 +3001,9 @@ Uri? _buildDomainUrl(String domain) {
 Future<List<MeshMessage>> fetchMessages({
   http.Client? client,
   String domain = 'potatomesh.net',
-  int limit = 1000,
+  int since = 0,
 }) async {
-  final uri = _buildMessagesUri(domain, limit: limit);
+  final uri = _buildMessagesUri(domain, since: since, limit: 1000);
   _logHttp('GET $uri');
 
   final httpClient = client ?? http.Client();

app/test/mesh_message_test.dart

@@ -168,6 +168,7 @@ void main() {
       final messages = await fetchMessages(client: client);
 
       expect(calls.single.queryParameters['limit'], '1000');
+      expect(calls.single.queryParameters['since'], '0');
       expect(messages.first.id, 1);
       expect(messages.last.id, 2);
       expect(messages.first.fromShort, 'a');

app/test/mesh_repository_test.dart

@@ -125,7 +125,7 @@ void main() {
   test('loadMessages performs incremental refresh after initial sync',
       () async {
     final nowSeconds = DateTime.now().toUtc().millisecondsSinceEpoch ~/ 1000;
-    final limits = <String>[];
+    final sinces = <String>[];
     final client = MockClient((request) async {
       if (request.url.path == '/api/nodes') {
         return http.Response(
@@ -143,8 +143,9 @@ void main() {
         );
       }
       if (request.url.path == '/api/messages') {
-        limits.add(request.url.queryParameters['limit'] ?? '');
-        if (limits.length == 1) {
+        sinces.add(request.url.queryParameters['since'] ?? '');
+        expect(request.url.queryParameters['limit'], '1000');
+        if (sinces.length == 1) {
           return http.Response(
             jsonEncode([
               {
@@ -187,8 +188,11 @@ void main() {
 
     final refreshed = await repository.loadMessages(domain: 'potatomesh.net');
 
-    expect(limits.first, '1000');
-    expect(limits.last, '100');
+    final expectedSince =
+        DateTime.parse('2024-01-01T00:00:00Z').toUtc().millisecondsSinceEpoch ~/
+            1000;
+    expect(sinces.first, '0');
+    expect(sinces.last, expectedSince.toString());
     expect(refreshed.length, 2);
     expect(refreshed.last.text, 'new');
   });

web/lib/potato_mesh/application/queries.rb

@@ -245,8 +245,17 @@ module PotatoMesh
         db&.close
       end
 
-      def query_messages(limit, node_ref: nil, include_encrypted: false)
+      # Fetch chat messages with optional filtering.
+      #
+      # @param limit [Integer] maximum number of rows to return.
+      # @param node_ref [String, Integer, nil] optional node reference to scope results.
+      # @param include_encrypted [Boolean] when true, include encrypted payloads in the response.
+      # @param since [Integer] unix timestamp threshold; messages with rx_time older than this are excluded.
+      # @return [Array<Hash>] compacted message rows safe for API responses.
+      def query_messages(limit, node_ref: nil, include_encrypted: false, since: 0)
         limit = coerce_query_limit(limit)
+        since_threshold = coerce_integer(since)
+        since_threshold = 0 if since_threshold.nil? || since_threshold.negative?
         db = open_database(readonly: true)
         db.results_as_hash = true
         params = []
@@ -254,10 +263,8 @@ module PotatoMesh
           "(COALESCE(TRIM(m.text), '') != '' OR COALESCE(TRIM(m.encrypted), '') != '' OR m.reply_id IS NOT NULL OR COALESCE(TRIM(m.emoji), '') != '')",
         ]
         include_encrypted = !!include_encrypted
-        now = Time.now.to_i
-        min_rx_time = now - PotatoMesh::Config.week_seconds
         where_clauses << "m.rx_time >= ?"
-        params << min_rx_time
+        params << since_threshold
 
         unless include_encrypted
           where_clauses << "COALESCE(TRIM(m.encrypted), '') = ''"

web/lib/potato_mesh/application/routes/api.rb

@@ -81,7 +81,9 @@ module PotatoMesh
             content_type :json
             limit = [params["limit"]&.to_i || 200, 1000].min
             include_encrypted = coerce_boolean(params["encrypted"]) || false
-            query_messages(limit, include_encrypted: include_encrypted).to_json
+            since = coerce_integer(params["since"])
+            since = 0 if since.nil? || since.negative?
+            query_messages(limit, include_encrypted: include_encrypted, since: since).to_json
           end
 
           app.get "/api/messages/:id" do
@@ -90,7 +92,14 @@ module PotatoMesh
             halt 400, { error: "missing node id" }.to_json unless node_ref
             limit = [params["limit"]&.to_i || 200, 1000].min
             include_encrypted = coerce_boolean(params["encrypted"]) || false
-            query_messages(limit, node_ref: node_ref, include_encrypted: include_encrypted).to_json
+            since = coerce_integer(params["since"])
+            since = 0 if since.nil? || since.negative?
+            query_messages(
+              limit,
+              node_ref: node_ref,
+              include_encrypted: include_encrypted,
+              since: since,
+            ).to_json
           end
 
           app.get "/api/positions" do

web/spec/app_spec.rb

@@ -3903,7 +3903,7 @@ RSpec.describe "Potato Mesh Sinatra app" do
       end
     end
 
-    it "omits messages received more than seven days ago" do
+    it "filters messages by the since parameter while defaulting to the full history" do
       clear_database
       allow(Time).to receive(:now).and_return(reference_time)
       now = reference_time.to_i
@@ -3930,14 +3930,19 @@ RSpec.describe "Potato Mesh Sinatra app" do
       expect(last_response).to be_ok
       payload = JSON.parse(last_response.body)
       ids = payload.map { |row| row["id"] }
-      expect(ids).to include(2)
-      expect(ids).not_to include(1)
+      expect(ids).to eq([2, 1])
 
-      get "/api/messages/!old"
+      get "/api/messages?since=#{fresh_rx}"
 
       expect(last_response).to be_ok
       filtered = JSON.parse(last_response.body)
       expect(filtered.map { |row| row["id"] }).to eq([2])
+
+      get "/api/messages/!old?since=#{fresh_rx}"
+
+      expect(last_response).to be_ok
+      scoped = JSON.parse(last_response.body)
+      expect(scoped.map { |row| row["id"] }).to eq([2])
     end
   end