@@ -3,8 +3,9 @@ package com.parseable.android.ui.screens.logviewer
33import androidx.lifecycle.ViewModel
44import androidx.lifecycle.viewModelScope
55import com.parseable.android.data.escapeIdentifier
6+import com.parseable.android.data.escapeLikePattern
67import com.parseable.android.data.escapeSql
7−import com.parseable.android.data.model.ApiResult
8+import com.parseable.android.data.model.*
89import com.parseable.android.data.repository.ParseableRepository
910import dagger.hilt.android.lifecycle.HiltViewModel
1011import kotlinx.coroutines.Job
@@ -15,8 +16,13 @@ import kotlinx.coroutines.flow.asStateFlow
1516import kotlinx.coroutines.flow.update
1617import kotlinx.coroutines.isActive
1718import kotlinx.coroutines.launch
19+import kotlinx.coroutines.sync.Mutex
20+import kotlinx.coroutines.sync.withLock
1821import kotlinx.serialization.json.JsonObject
1922import kotlinx.serialization.json.jsonPrimitive
23+import java.time.Instant
24+import java.time.LocalDateTime
25+import java.time.OffsetDateTime
2026import java.time.ZoneOffset
2127import java.time.ZonedDateTime
2228import java.time.format.DateTimeFormatter
@@ -34,10 +40,21 @@ enum class TimeRange(val label: String, val minutes: Long) {
3440 CUSTOM("Custom", 0),
3541}
3642
43+/** The original (column, operator, value) a filter chip was built from. */
44+data class FilterCondition(
45+ val column: String,
46+ val operator: String,
47+ val value: String,
48+)
49+
3750data class FilterState(
3851 val searchQuery: String = "",
3952 val activeFilters: List<String> = emptyList(),
4053 val filterClauses: List<String> = emptyList(),
54+ // Structured form of each chip, index-aligned with activeFilters/filterClauses.
55+ // Kept so saved filters reconstruct from real values instead of re-parsing the
56+ // human-readable display string (which is ambiguous when a value contains an operator).
57+ val filterConditions: List<FilterCondition> = emptyList(),
4158 val customSql: String = "",
4259 val isSearching: Boolean = false,
4360)
@@ -49,19 +66,33 @@ data class StreamingState(
4966 val currentIntervalMs: Long = 3000L,
5067)
5168
69+data class SavedFiltersState(
70+ val filters: List<SavedFilter> = emptyList(),
71+ val isLoading: Boolean = false,
72+ val isSaving: Boolean = false,
73+ val error: String? = null,
74+)
75+
5276data class LogViewerState(
5377 val streamName: String = "",
5478 val logs: List<JsonObject> = emptyList(),
79+ val logKeys: List<String> = emptyList(),
5580 val columns: List<String> = emptyList(),
81+ val searchableColumns: List<String> = emptyList(),
5682 val isLoading: Boolean = false,
5783 val error: String? = null,
5884 val selectedTimeRange: TimeRange = TimeRange.LAST_1H,
5985 val filters: FilterState = FilterState(),
6086 val streaming: StreamingState = StreamingState(),
87+ val savedFilters: SavedFiltersState = SavedFiltersState(),
6188 val currentLimit: Int = 500,
6289 val hasMore: Boolean = false,
6390 val customStartTime: Long? = null,
6491 val customEndTime: Long? = null,
92+ // Upper time bound (epoch millis) fixed at the first page of a paging session, so loadMore()
93+ // doesn't slide a relative ("last N minutes") window forward as it re-queries. Null until the
94+ // first refresh of a session; re-anchored whenever currentLimit resets to the initial page.
95+ val pageAnchorTime: Long? = null,
6596) {
6697 // Convenience accessors for backward compatibility with Screen
6798 val searchQuery: String get() = filters.searchQuery
@@ -89,17 +120,53 @@ class LogViewerViewModel @Inject constructor(
89120 @Volatile private var streamingGeneration: Int = 0
90121 @Volatile private var lastSeenTimestamp: String? = null
91122 @Volatile private var consecutiveStreamingErrors: Int = 0
123+ // True while the log list is scrolled to the top. New rows prepend at the top, so when the
124+ // user is already there they see them live and the "+N new" badge must stay at 0. The badge
125+ // only accumulates while the user is scrolled away, and resets when they return to the top.
126+ @Volatile private var viewingTop: Boolean = true
127+
128+ // Identities of rows already surfaced during the current streaming session. Used to dedupe
129+ // re-fetched boundary rows. Kept INDEPENDENT of the displayed `logs` list — which is capped
130+ // at STREAMING_MAX_LOGS — so that under sustained high-volume streaming, boundary rows
131+ // evicted from the view are still recognized as duplicates and not prepended twice.
132+ // Insertion-ordered so the oldest keys are trimmed first. Touched only from the single
133+ // streaming coroutine and from startStreaming (which first cancels that coroutine).
134+ private val seenStreamingKeys = LinkedHashSet<String>()
92135 private var searchJob: Job? = null
93136 private var schemaJob: Job? = null
137+ private var refreshJob: Job? = null
94138
95139 companion object {
96140 private const val STREAMING_BASE_INTERVAL_MS = 3000L
97141 private const val STREAMING_MAX_INTERVAL_MS = 30000L
98142 private const val STREAMING_MAX_LOGS = 1000
143+ // Remember more identities than the display cap so a row evicted from the view is still
144+ // deduped on a later re-fetch. 2x the display cap covers the worst realistic overlap.
145+ private const val SEEN_STREAMING_KEYS_MAX = STREAMING_MAX_LOGS * 2
99146 private const val MAX_LOAD_LIMIT = 5000
100147 private const val LOAD_MORE_INCREMENT = 500
101148 private const val MAX_STREAMING_ERRORS = 5
102149 private val ALLOWED_OPERATORS = setOf("=", "!=", "LIKE", "ILIKE", ">", "<", ">=", "<=", "IS NULL", "IS NOT NULL")
150+
151+ /**
152+ * Produces deterministic keys for a list of log entries so that expanded state
153+ * survives list mutations (e.g., new logs prepended during streaming).
154+ */
155+ internal fun computeLogKeys(logs: List<JsonObject>): List<String> {
156+ val seen = mutableMapOf<String, Int>()
157+ return logs.map { log ->
158+ // Key on the full row content (same identity used for streaming dedup, see
159+ // logIdentity). Keying on only p_timestamp/p_metadata/p_tags collided for rows
160+ // sharing those fields but differing in body, so the positional tiebreaker made
161+ // keys shift when streaming prepended rows — corrupting expanded state and
162+ // letting remember(stableKey) blocks render a stale body. The trailing counter
163+ // only disambiguates genuinely identical rows, whose rendering is identical too.
164+ val base = log.toString()
165+ val count = seen.getOrDefault(base, 0)
166+ seen[base] = count + 1
167+ if (count == 0) base else "$base|$count"
168+ }
169+ }
103170 }
104171
105172 fun initialize(streamName: String) {
@@ -115,8 +182,9 @@ class LogViewerViewModel @Inject constructor(
115182 schemaJob = viewModelScope.launch {
116183 when (val result = repository.getStreamSchema(streamName)) {
117184 is ApiResult.Success -> {
118− val columns = result.data.fields.map { it.name }
119− _state.update { it.copy(columns = columns) }
185+ val columns = result.data.fields.map { it.name }.sorted()
186+ val searchable = columns.filter { !it.startsWith("p_") }
187+ _state.update { it.copy(columns = columns, searchableColumns = searchable) }
120188 }
121189 is ApiResult.Error -> { /* Schema loading failure is non-fatal */ }
122190 }
@@ -150,7 +218,11 @@ class LogViewerViewModel @Inject constructor(
150218 }
151219
152220 fun onSearchQueryChange(query: String) {
153− _state.update { it.copy(filters = it.filters.copy(searchQuery = query, isSearching = query.isNotBlank())) }
221+ // A new query invalidates the live-tail baseline and the paged-up limit, matching
222+ // the time-range / custom-SQL handlers.
223+ stopStreaming()
224+ // Searching returns to the filter-builder query path, so drop any active custom SQL.
225+ _state.update { it.copy(currentLimit = 500, filters = it.filters.copy(searchQuery = query, customSql = "", isSearching = query.isNotBlank())) }
154226 searchJob?.cancel()
155227 searchJob = viewModelScope.launch {
156228 delay(300)
@@ -159,28 +231,55 @@ class LogViewerViewModel @Inject constructor(
159231 }
160232 }
161233
234+ /**
235+ * Run the current search immediately (e.g. the IME "Search" action). Cancels the pending
236+ * debounce so it can't fire a second, redundant refresh ~300ms later.
237+ */
238+ fun submitSearch() {
239+ searchJob?.cancel()
240+ _state.update { it.copy(filters = it.filters.copy(isSearching = false)) }
241+ stopStreaming()
242+ refresh()
243+ }
244+
162245 fun addFilter(column: String, operator: String, value: String) {
246+ addFilterInternal(column, operator, value)
247+ refresh()
248+ }
249+
250+ private fun addFilterInternal(column: String, operator: String, value: String) {
163251 if (operator !in ALLOWED_OPERATORS) return
164252 val safeColumn = escapeIdentifier(column)
165− val safeValue = escapeSql(value)
166253 val clause = when (operator) {
167254 "IS NULL", "IS NOT NULL" -> "\"$safeColumn\" $operator"
168− "LIKE", "ILIKE" -> "\"$safeColumn\" $operator '%$safeValue%'"
169− else -> "\"$safeColumn\" $operator '$safeValue'"
255+ // %/_ in the user's value must match literally, so escape the LIKE wildcards
256+ // and declare the escape character. Exact comparisons keep %/_ verbatim.
257+ "LIKE", "ILIKE" -> "\"$safeColumn\" $operator '%${escapeLikePattern(value)}%' ESCAPE '\\'"
258+ else -> "\"$safeColumn\" $operator '${escapeSql(value)}'"
259+ }
260+ // IS NULL / IS NOT NULL ignore the value in both the clause and the display, so store
261+ // an empty value in the structured condition too (keeps saved filters clean).
262+ val conditionValue = when (operator) {
263+ "IS NULL", "IS NOT NULL" -> ""
264+ else -> value
170265 }
171266 val display = when (operator) {
172267 "IS NULL", "IS NOT NULL" -> "$column $operator"
173268 else -> "$column $operator $value"
174269 }
270+ // A new filter invalidates the live-tail baseline and the paged-up limit.
271+ stopStreaming()
175272 _state.update {
176273 it.copy(
274+ currentLimit = 500,
177275 filters = it.filters.copy(
178276 filterClauses = it.filters.filterClauses + clause,
179277 activeFilters = it.filters.activeFilters + display,
278+ filterConditions = it.filters.filterConditions + FilterCondition(column, operator, conditionValue),
279+ customSql = "",
180280 ),
181281 )
182282 }
183− refresh()
184283 }
185284
186285 fun removeFilter(display: String) {
@@ -190,81 +289,93 @@ class LogViewerViewModel @Inject constructor(
190289 if (index >= 0 && index < it.filters.filterClauses.size) {
191290 removed = true
192291 it.copy(
292+ currentLimit = 500,
193293 filters = it.filters.copy(
194294 activeFilters = it.filters.activeFilters.filterIndexed { i, _ -> i != index },
195295 filterClauses = it.filters.filterClauses.filterIndexed { i, _ -> i != index },
296+ filterConditions = it.filters.filterConditions.filterIndexed { i, _ -> i != index },
297+ customSql = "",
196298 ),
197299 )
198300 } else {
199301 it
200302 }
201303 }
202− if (removed) refresh()
304+ if (removed) {
305+ // A changed filter set invalidates the live-tail baseline.
306+ stopStreaming()
307+ refresh()
308+ }
203309 }
204310
205311 fun clearFilters() {
312+ stopStreaming()
206313 _state.update {
207− it.copy(filters = it.filters.copy(activeFilters = emptyList(), filterClauses = emptyList()))
314+ it.copy(currentLimit = 500, filters = it.filters.copy(activeFilters = emptyList(), filterClauses = emptyList(), filterConditions = emptyList(), customSql = "", searchQuery = "", isSearching = false))
208315 }
209316 refresh()
210317 }
211318
212319 fun executeCustomSql(sql: String) {
213− val trimmed = sql.trim()
320+ // Custom SQL changes the query basis, so cancel any pending search debounce that would
321+ // otherwise fire a redundant refresh() ~300ms later. (Unlike the filter/time-range
322+ // handlers, this path doesn't route through stopStreaming().)
323+ searchJob?.cancel()
324+ // Drop a trailing statement terminator before we append anything — otherwise
325+ // "SELECT * FROM x;" would become "SELECT * FROM x; LIMIT 5000", a syntax error.
326+ val trimmed = sql.trim().trimEnd(';').trim()
214327 if (!trimmed.uppercase().startsWith("SELECT")) {
215328 _state.update { it.copy(error = "Only SELECT queries are allowed") }
216329 return
217330 }
218− // Enforce a LIMIT to prevent OOM from unbounded queries
219− val safeSql = if (!trimmed.uppercase().contains("LIMIT")) {
331+ // Enforce a LIMIT to prevent OOM from unbounded queries. Only honor a *top-level*
332+ // `LIMIT <n>` (optionally with OFFSET) anchored at the end of the statement — a
333+ // LIMIT inside a subquery doesn't bound the outer result set, so matching it
334+ // anywhere would let an unbounded outer query slip past the safety limit. Anchoring
335+ // at the end also means a string literal like 'LIMIT EXCEEDED' can't disable it.
336+ val hasTopLevelLimit =
337+ Regex("(?i)\\bLIMIT\\s+\\d+(\\s+OFFSET\\s+\\d+)?\\s*$").containsMatchIn(trimmed)
338+ val safeSql = if (!hasTopLevelLimit) {
220339 "$trimmed LIMIT $MAX_LOAD_LIMIT"
221340 } else {
222341 trimmed
223342 }
343+ // Store the custom SQL and let refresh() run it. Keeping it in state means
344+ // pull-to-refresh / time-range changes re-run the custom query instead of
345+ // silently replacing it with the default filter query.
224346 _state.update { it.copy(filters = it.filters.copy(customSql = safeSql), currentLimit = 500) }
225− viewModelScope.launch {
226− _state.update { it.copy(isLoading = true, error = null) }
227−
228− val (startTime, endTime) = getTimeRange()
229− when (val result = repository.queryLogsRaw(safeSql, startTime, endTime)) {
230− is ApiResult.Success -> {
231− _state.update {
232− it.copy(
233− logs = result.data,
234− isLoading = false,
235− hasMore = result.data.size >= it.currentLimit,
236− )
237− }
238− }
239− is ApiResult.Error -> {
240− _state.update {
241− it.copy(isLoading = false, error = result.userMessage)
242− }
243− }
244− }
245− }
347+ refresh()
246348 }
247349
248− private fun buildSearchClause(columns: List<String>, searchQuery: String): String? {
350+ private fun buildSearchClause(searchableColumns: List<String>, searchQuery: String): String? {
249351 if (searchQuery.isBlank()) return null
250− val safeSearch = escapeSql(searchQuery)
251− // Search across ALL non-internal columns (not just 5)
252− val searchFields = columns.filter { !it.startsWith("p_") }
253− if (searchFields.isEmpty()) return null
254− return searchFields.joinToString(" OR ") {
255− "\"${escapeIdentifier(it)}\" ILIKE '%$safeSearch%'"
352+ if (searchableColumns.isEmpty()) return null
353+ val safeSearch = escapeLikePattern(searchQuery)
354+ return searchableColumns.joinToString(" OR ") {
355+ "CAST(\"${escapeIdentifier(it)}\" AS VARCHAR) ILIKE '%$safeSearch%' ESCAPE '\\'"
256356 }
257357 }
258358
259359 fun refresh() {
260360 if (_state.value.streamName.isEmpty()) return
261361
262− viewModelScope.launch {
362+ // Cancel any in-flight query so out-of-order completions can't overwrite
363+ // fresh results with stale ones (filter/search/time-range/loadMore all call refresh()).
364+ refreshJob?.cancel()
365+ refreshJob = viewModelScope.launch {
366+ // Anchor the upper time bound on the first page of a paging session (currentLimit at
367+ // its initial value) and reuse it for subsequent loadMore() pages. Otherwise a relative
368+ // range would recompute "now" on every page in getTimeRange(), sliding the window and
369+ // duplicating/dropping rows on an actively-ingesting stream.
370+ val anchor = _state.value.let { s ->
371+ if (s.currentLimit <= 500) System.currentTimeMillis() else s.pageAnchorTime
372+ }
263373 _state.update {
264374 it.copy(
265375 isLoading = true,
266376 error = null,
267377 streaming = it.streaming.copy(streamingError = null),
378+ pageAnchorTime = anchor,
268379 )
269380 }
270381
@@ -272,10 +383,24 @@ class LogViewerViewModel @Inject constructor(
272383 val current = _state.value
273384 val (startTime, endTime) = getTimeRange()
274385
386+ // A custom SQL query owns the whole statement (its own LIMIT, columns, ordering),
387+ // so re-run it verbatim. It is not paginated via loadMore(), hence hasMore = false.
388+ if (current.filters.customSql.isNotBlank()) {
389+ when (val result = repository.queryLogsRaw(current.filters.customSql, startTime, endTime)) {
390+ is ApiResult.Success -> {
391+ _state.update { it.copy(logs = result.data, logKeys = computeLogKeys(result.data), isLoading = false, hasMore = false) }
392+ }
393+ is ApiResult.Error -> {
394+ _state.update { it.copy(isLoading = false, error = result.userMessage) }
395+ }
396+ }
397+ return@launch
398+ }
399+
275400 // Build WHERE clause from filters + search
276401 val clauses = current.filters.filterClauses.toMutableList()
277402 if (current.filters.searchQuery.isNotBlank()) {
278− val searchClause = buildSearchClause(current.columns, current.filters.searchQuery)
403+ val searchClause = buildSearchClause(current.searchableColumns, current.filters.searchQuery)
279404 if (searchClause != null) {
280405 clauses.add("($searchClause)")
281406 } else {
@@ -288,20 +413,26 @@ class LogViewerViewModel @Inject constructor(
288413
289414 val filterSql = clauses.joinToString(" AND ")
290415
416+ // Over-fetch by one row so we can distinguish "exactly currentLimit rows exist"
417+ // (no more pages) from "the page is full and more may exist" without firing a
418+ // wasted extra query the next time the user scrolls to the bottom.
419+ val requestLimit = current.currentLimit + 1
291420 when (val result = repository.queryLogs(
292421 stream = current.streamName,
293422 startTime = startTime,
294423 endTime = endTime,
295424 filterSql = filterSql,
296− limit = current.currentLimit,
425+ limit = requestLimit,
297426 )) {
298427 is ApiResult.Success -> {
428+ val hasMore = result.data.size > current.currentLimit && current.currentLimit < MAX_LOAD_LIMIT
429+ val logs = if (result.data.size > current.currentLimit) {
430+ result.data.take(current.currentLimit)
431+ } else {
432+ result.data
433+ }
299434 _state.update {
300− it.copy(
301− logs = result.data,
302− isLoading = false,
303− hasMore = result.data.size >= it.currentLimit,
304− )
435+ it.copy(logs = logs, logKeys = computeLogKeys(logs), isLoading = false, hasMore = hasMore)
305436 }
306437 }
307438 is ApiResult.Error -> {
@@ -313,8 +444,29 @@ class LogViewerViewModel @Inject constructor(
313444 }
314445 }
315446
447+ /**
448+ * Pull-to-refresh: restart the paging session from the first page. Resetting currentLimit
449+ * makes refresh() re-anchor pageAnchorTime to "now", so a relative time range advances and
450+ * newly-ingested rows appear. Calling refresh() alone keeps the frozen anchor once the user
451+ * has paged past the first page, making pull-to-refresh look like it does nothing.
452+ */
453+ fun pullRefresh() {
454+ if (_state.value.streamName.isEmpty()) return
455+ // While live-tailing, the poller continuously prepends fresh rows; a concurrent refresh()
456+ // would replace the list with a full page at the same time, racing the poller on logs/
457+ // logKeys/streamingError and producing duplicated or dropped rows plus a stale "+N" badge.
458+ // The displayed logs are already current under streaming, so the pull is a no-op here —
459+ // mirror loadMore()'s guard rather than the stopStreaming() path the other mutators take.
460+ if (_state.value.streaming.isStreaming) return
461+ _state.update { it.copy(currentLimit = 500) }
462+ refresh()
463+ }
464+
316465 fun loadMore() {
317466 val current = _state.value
467+ // While live-tailing, logs are prepended in memory; a refresh() here would replace
468+ // them with a fresh page and drop the streamed rows. Pagination resumes once stopped.
469+ if (current.streaming.isStreaming) return
318470 if (current.currentLimit >= MAX_LOAD_LIMIT) {
319471 _state.update { it.copy(hasMore = false) }
320472 return
@@ -337,7 +489,9 @@ class LogViewerViewModel @Inject constructor(
337489 )
338490 return Pair(start.format(dateFormatter), end.format(dateFormatter))
339491 }
340− val now = ZonedDateTime.now(ZoneOffset.UTC)
492+ val now = current.pageAnchorTime
493+ ?.let { ZonedDateTime.ofInstant(java.time.Instant.ofEpochMilli(it), ZoneOffset.UTC) }
494+ ?: ZonedDateTime.now(ZoneOffset.UTC)
341495 val start = now.minusMinutes(current.selectedTimeRange.minutes)
342496 return Pair(
343497 start.format(dateFormatter),
@@ -345,6 +499,17 @@ class LogViewerViewModel @Inject constructor(
345499 )
346500 }
347501
502+ /**
503+ * Report whether the log list is scrolled to the top. Clears the "+N new" streaming badge
504+ * when the user returns to the top, since prepended rows there are seen immediately.
505+ */
506+ fun setViewingTop(atTop: Boolean) {
507+ viewingTop = atTop
508+ if (atTop && _state.value.streaming.streamingNewCount != 0) {
509+ _state.update { it.copy(streaming = it.streaming.copy(streamingNewCount = 0)) }
510+ }
511+ }
512+
348513 fun toggleStreaming() {
349514 if (_state.value.streaming.isStreaming) {
350515 stopStreaming()
@@ -354,6 +519,10 @@ class LogViewerViewModel @Inject constructor(
354519 }
355520
356521 private fun startStreaming() {
522+ // The poller builds its own default SELECT * query and cannot honor a custom SQL
523+ // projection/filter, so streaming custom-query results would prepend mismatched rows.
524+ // The UI disables the toggle in this state; guard here too in case it's reached anyway.
525+ if (_state.value.filters.customSql.isNotBlank()) return
357526 stopStreaming()
358527 consecutiveStreamingErrors = 0
359528 val generation = ++streamingGeneration
@@ -361,6 +530,12 @@ class LogViewerViewModel @Inject constructor(
361530 it.copy(streaming = StreamingState(isStreaming = true, streamingNewCount = 0))
362531 }
363532
533+ // Seed the dedup set with the rows already on screen so an initial row sitting exactly
534+ // on the streaming baseline timestamp can't be re-fetched and duplicated.
535+ seenStreamingKeys.clear()
536+ _state.value.logs.forEach { seenStreamingKeys.add(logIdentity(it)) }
537+ trimSeenStreamingKeys()
538+
364539 // Set the baseline timestamp to "now" so we only poll for new logs
365540 lastSeenTimestamp = ZonedDateTime.now(ZoneOffset.UTC).format(dateFormatter)
366541
@@ -383,6 +558,13 @@ class LogViewerViewModel @Inject constructor(
383558 }
384559
385560 fun stopStreaming() {
561+ // Cancel any pending search debounce too. Every query-basis change (time range, filters,
562+ // saved filter, clear, new search) routes through stopStreaming(); without this, a
563+ // searchJob scheduled moments before such a change would still fire refresh() ~300ms
564+ // later and redundantly re-query over the just-applied state. onSearchQueryChange()
565+ // re-assigns searchJob immediately after calling stopStreaming(), so its own debounce
566+ // survives.
567+ searchJob?.cancel()
386568 streamingGeneration++
387569 streamingJob?.cancel()
388570 streamingJob = null
@@ -397,6 +579,49 @@ class LogViewerViewModel @Inject constructor(
397579 }
398580 }
399581
582+ /**
583+ * Content-based identity for a log row, used to dedupe re-fetched boundary logs.
584+ * Uses the full row: two rows are duplicates only when every field matches, so
585+ * distinct logs that happen to share a timestamp/metadata/tags are not dropped.
586+ */
587+ private fun logIdentity(log: JsonObject): String = log.toString()
588+
589+ /**
590+ * Parse a server-rendered p_timestamp (which may be ISO with 'Z', offset-carrying,
591+ * space-separated, or offset-less UTC) and reformat it to the canonical query format
592+ * (`dateFormatter`) so it's a valid inclusive startTime for the next poll. Returns null
593+ * if the value can't be parsed, so the caller can keep the previous boundary.
594+ */
595+ private fun normalizeBoundaryTimestamp(raw: String): String? {
596+ val isoLike = if (' ' in raw && 'T' !in raw) raw.replaceFirst(' ', 'T') else raw
597+ val instant = try {
598+ Instant.parse(isoLike)
599+ } catch (_: Exception) {
600+ try {
601+ OffsetDateTime.parse(isoLike).toInstant()
602+ } catch (_: Exception) {
603+ try {
604+ LocalDateTime.parse(isoLike).toInstant(ZoneOffset.UTC)
605+ } catch (_: Exception) {
606+ return null
607+ }
608+ }
609+ }
610+ return instant.atOffset(ZoneOffset.UTC).format(dateFormatter)
611+ }
612+
613+ /** Bound the dedup set, evicting the oldest identities first (insertion order). */
614+ private fun trimSeenStreamingKeys() {
615+ if (seenStreamingKeys.size <= SEEN_STREAMING_KEYS_MAX) return
616+ val iterator = seenStreamingKeys.iterator()
617+ var toRemove = seenStreamingKeys.size - SEEN_STREAMING_KEYS_MAX
618+ while (toRemove > 0 && iterator.hasNext()) {
619+ iterator.next()
620+ iterator.remove()
621+ toRemove--
622+ }
623+ }
624+
400625 private suspend fun pollNewLogs() {
401626 val current = _state.value
402627 if (current.streamName.isEmpty()) return
@@ -407,46 +632,118 @@ class LogViewerViewModel @Inject constructor(
407632 // Build WHERE clause from active filters + search
408633 val clauses = current.filters.filterClauses.toMutableList()
409634 if (current.filters.searchQuery.isNotBlank()) {
410− val searchClause = buildSearchClause(current.columns, current.filters.searchQuery)
635+ val searchClause = buildSearchClause(current.searchableColumns, current.filters.searchQuery)
411636 if (searchClause != null) {
412− clauses.add(searchClause)
637+ // Parenthesize: the search clause is OR-joined and must not break the
638+ // precedence of the AND-joined filter clauses around it.
639+ clauses.add("($searchClause)")
640+ } else {
641+ // Schema not loaded yet, so we can't build the search filter. Skip this
642+ // poll rather than querying without it — otherwise we'd prepend rows that
643+ // don't match the active search into a filtered view. Mirrors refresh(),
644+ // which errors out instead of running an unfiltered query.
645+ _state.update {
646+ it.copy(
647+ streaming = it.streaming.copy(
648+ streamingError = "Search unavailable: stream schema not loaded",
649+ ),
650+ )
651+ }
652+ return
413653 }
414654 }
415655
416656 val whereClause = if (clauses.isNotEmpty()) " WHERE ${clauses.joinToString(" AND ")}" else ""
417657 val safeName = escapeIdentifier(current.streamName)
418− val sql = "SELECT * FROM \"$safeName\"$whereClause ORDER BY p_timestamp DESC LIMIT 200"
658+ // Fetch up to the in-memory display cap. With a small limit (e.g. 200), a burst of
659+ // more than that many logs in one interval would advance lastSeenTimestamp past the
660+ // un-fetched rows and drop them permanently. Capping at STREAMING_MAX_LOGS means we
661+ // never skip rows we'd actually keep (anything older is evicted by the cap anyway).
662+ val sql = "SELECT * FROM \"$safeName\"$whereClause ORDER BY p_timestamp DESC LIMIT $STREAMING_MAX_LOGS"
419663
420664 when (val result = repository.queryLogsRaw(sql, startTime, endTime)) {
421665 is ApiResult.Success -> {
422666 val newLogs = result.data
423667 if (newLogs.isNotEmpty()) {
424− // Update the last seen timestamp to the most recent log
425− val newestTimestamp = try {
426− newLogs.firstOrNull()
427− ?.get("p_timestamp")?.jsonPrimitive?.content
668+ // Advance the last-seen boundary. Rows are ordered newest-first, so
669+ // normally we jump to the newest. But a full page means the LIMIT may
670+ // have capped off matching rows OLDER than the newest in this batch
671+ // (possible once filters/search narrow a high-volume stream) — advancing
672+ // to the newest would skip them forever. In that case advance only to the
673+ // oldest row we did fetch, so the next poll re-queries forward and closes
674+ // the gap (re-fetched rows are deduped below).
675+ val fullPage = newLogs.size >= STREAMING_MAX_LOGS
676+ val boundaryLog = if (fullPage) newLogs.lastOrNull() else newLogs.firstOrNull()
677+ val boundaryTimestamp = try {
678+ boundaryLog?.get("p_timestamp")?.jsonPrimitive?.content
428679 } catch (_: Exception) {
429680 null
430681 }
431− if (newestTimestamp != null) {
432− lastSeenTimestamp = newestTimestamp
682+ // Re-canonicalize the server's p_timestamp before reusing it as the next
683+ // query's startTime. The raw value can be space-separated or offset-less,
684+ // which the query API may reject or misparse — that would stall live-tail
685+ // after the first batch. If it can't be parsed, leave lastSeenTimestamp as-is
686+ // (the inclusive startTime simply re-fetches the window; rows are deduped).
687+ var normalized = boundaryTimestamp?.let { normalizeBoundaryTimestamp(it) }
688+ // On a saturated (full) page the boundary is the OLDEST fetched row so the
689+ // next poll re-queries forward and closes the LIMIT gap. But if that oldest
690+ // row carries the SAME timestamp we're already at — e.g. a burst of more than
691+ // STREAMING_MAX_LOGS rows sharing one millisecond, or the parse failing — the
692+ // boundary wouldn't move and we'd re-query the identical window forever. Fall
693+ // back to the NEWEST row's timestamp to guarantee forward progress; the
694+ // skipped backlog would be evicted by the display cap anyway.
695+ if (fullPage && (normalized == null || normalized == lastSeenTimestamp)) {
696+ val newestTimestamp = try {
697+ newLogs.firstOrNull()?.get("p_timestamp")?.jsonPrimitive?.content
698+ } catch (_: Exception) {
699+ null
700+ }
701+ val newestNormalized = newestTimestamp?.let { normalizeBoundaryTimestamp(it) }
702+ if (newestNormalized != null && newestNormalized != lastSeenTimestamp) {
703+ normalized = newestNormalized
704+ }
705+ }
706+ if (normalized != null) {
707+ lastSeenTimestamp = normalized
433708 }
434709
435− _state.update { state ->
436− // Prepend new logs, cap total
437− val combined = newLogs + state.logs
438− val capped = if (combined.size > STREAMING_MAX_LOGS) {
439− combined.take(STREAMING_MAX_LOGS)
440− } else {
441− combined
710+ // The query's startTime is inclusive, so rows sitting on the previous
711+ // boundary timestamp get re-fetched. Dedup against everything seen this
712+ // session (independent of the display cap) to avoid duplicate rows and an
713+ // inflated count. Computed BEFORE the state update — and the seen-set
714+ // mutated here, not inside the update lambda, which may re-run.
715+ val freshLogs = newLogs.filterNot { logIdentity(it) in seenStreamingKeys }
716+ newLogs.forEach { seenStreamingKeys.add(logIdentity(it)) }
717+ trimSeenStreamingKeys()
718+
719+ if (freshLogs.isEmpty()) {
720+ _state.update { it.copy(streaming = it.streaming.copy(streamingError = null)) }
721+ } else {
722+ _state.update { state ->
723+ // Prepend new logs, cap total — avoid full intermediate list allocation
724+ val remaining = (STREAMING_MAX_LOGS - freshLogs.size).coerceAtLeast(0)
725+ val capped = ArrayList<JsonObject>(freshLogs.size + remaining).apply {
726+ addAll(freshLogs)
727+ val oldLogs = state.logs
728+ addAll(oldLogs.subList(0, remaining.coerceAtMost(oldLogs.size)))
729+ }
730+ state.copy(
731+ logs = capped,
732+ logKeys = computeLogKeys(capped),
733+ streaming = state.streaming.copy(
734+ // Accumulate new rows while the user is scrolled away; reset
735+ // to 0 when they're at the top (where prepended rows are seen
736+ // live). The previous cap was against total list size, so the
737+ // badge saturated at the display cap and never cleared.
738+ streamingNewCount = if (viewingTop) {
739+ 0
740+ } else {
741+ state.streaming.streamingNewCount + freshLogs.size
742+ },
743+ streamingError = null,
744+ ),
745+ )
442746 }
443− state.copy(
444− logs = capped,
445− streaming = state.streaming.copy(
446− streamingNewCount = state.streaming.streamingNewCount + newLogs.size,
447− streamingError = null,
448− ),
449− )
450747 }
451748 }
452749 consecutiveStreamingErrors = 0
@@ -471,6 +768,191 @@ class LogViewerViewModel @Inject constructor(
471768 }
472769 }
473770
771+ // --- Saved Filters (synced with Parseable server) ---
772+
773+ // Serializes loadSavedFilters / saveCurrentFilter / deleteSavedFilter so they never
774+ // interleave. Without this, a wholesale-overwrite load completing after a concurrent
775+ // save (append) or delete (remove) would silently drop the local add/remove — a
776+ // read-modify-write race across coroutines.
777+ private val savedFiltersMutex = Mutex()
778+
779+ fun loadSavedFilters() {
780+ // Capture the stream this load is for; if the user navigates to a different stream
781+ // before it completes, bind the result to the right stream instead of clobbering the
782+ // new stream's filter list with another stream's results.
783+ val stream = _state.value.streamName
784+ viewModelScope.launch {
785+ savedFiltersMutex.withLock {
786+ _state.update { it.copy(savedFilters = it.savedFilters.copy(isLoading = true, error = null)) }
787+ when (val result = repository.listFilters()) {
788+ is ApiResult.Success -> {
789+ if (_state.value.streamName != stream) {
790+ _state.update { it.copy(savedFilters = it.savedFilters.copy(isLoading = false)) }
791+ return@withLock
792+ }
793+ val streamFilters = result.data
794+ .filter { it.streamName == stream }
795+ .sortedBy { it.filterName.lowercase() }
796+ _state.update {
797+ it.copy(savedFilters = it.savedFilters.copy(filters = streamFilters, isLoading = false))
798+ }
799+ }
800+ is ApiResult.Error -> {
801+ _state.update {
802+ it.copy(savedFilters = it.savedFilters.copy(isLoading = false, error = result.userMessage))
803+ }
804+ }
805+ }
806+ }
807+ }
808+ }
809+
810+ fun saveCurrentFilter(name: String) {
811+ val current = _state.value
812+ if (current.streamName.isEmpty()) return
813+
814+ viewModelScope.launch {
815+ savedFiltersMutex.withLock {
816+ _state.update { it.copy(savedFilters = it.savedFilters.copy(isSaving = true, error = null)) }
817+
818+ val query = if (current.filters.customSql.isNotBlank()) {
819+ SavedFilterQuery(
820+ filterType = "sql",
821+ filterQuery = current.filters.customSql,
822+ )
823+ } else {
824+ val ruleGroups = current.filters.filterConditions
825+ .mapIndexed { index, condition ->
826+ FilterRule(
827+ id = "rule_$index",
828+ field = condition.column,
829+ value = condition.value,
830+ operator = condition.operator,
831+ )
832+ }
833+ val builder = if (ruleGroups.isNotEmpty()) {
834+ FilterBuilder(
835+ id = "root",
836+ combinator = "and",
837+ rules = listOf(
838+ FilterRuleGroup(
839+ id = "group_0",
840+ combinator = "and",
841+ rules = ruleGroups,
842+ ),
843+ ),
844+ )
845+ } else null
846+
847+ SavedFilterQuery(
848+ filterType = if (current.filters.searchQuery.isNotBlank()) "search" else "filter",
849+ filterQuery = current.filters.searchQuery.ifBlank { null },
850+ filterBuilder = builder,
851+ )
852+ }
853+
854+ val filter = SavedFilter(
855+ filterName = name,
856+ streamName = current.streamName,
857+ query = query,
858+ )
859+
860+ when (val result = repository.createFilter(filter)) {
861+ is ApiResult.Success -> {
862+ _state.update {
863+ it.copy(
864+ savedFilters = it.savedFilters.copy(
865+ isSaving = false,
866+ filters = it.savedFilters.filters + result.data,
867+ ),
868+ )
869+ }
870+ }
871+ is ApiResult.Error -> {
872+ _state.update {
873+ it.copy(savedFilters = it.savedFilters.copy(isSaving = false, error = result.userMessage))
874+ }
875+ }
876+ }
877+ }
878+ }
879+ }
880+
881+ fun applySavedFilter(filter: SavedFilter) {
882+ stopStreaming()
883+
884+ when (filter.query.filterType) {
885+ "sql" -> {
886+ val sql = filter.query.filterQuery ?: return
887+ // Reset to a clean filter state first, then let executeCustomSql validate
888+ // the SQL and own the customSql state. Pre-seeding filters.customSql with
889+ // the raw, unvalidated query meant that if executeCustomSql rejected it
890+ // (e.g. a legacy non-SELECT saved filter, which returns early without
891+ // touching state) we were left wedged in custom-SQL mode with broken SQL
892+ // that every later refresh() would keep re-running.
893+ _state.update {
894+ it.copy(
895+ filters = FilterState(),
896+ currentLimit = 500,
897+ )
898+ }
899+ executeCustomSql(sql)
900+ }
901+ "search" -> {
902+ val searchQuery = filter.query.filterQuery ?: ""
903+ _state.update {
904+ it.copy(
905+ filters = FilterState(searchQuery = searchQuery),
906+ currentLimit = 500,
907+ )
908+ }
909+ // Also apply builder rules if present
910+ applyBuilderRules(filter.query.filterBuilder)
911+ refresh()
912+ }
913+ else -> {
914+ _state.update {
915+ it.copy(
916+ filters = FilterState(),
917+ currentLimit = 500,
918+ )
919+ }
920+ applyBuilderRules(filter.query.filterBuilder)
921+ refresh()
922+ }
923+ }
924+ }
925+
926+ private fun applyBuilderRules(builder: FilterBuilder?) {
927+ if (builder == null) return
928+ for (group in builder.rules) {
929+ for (rule in group.rules) {
930+ if (rule.field.isNotBlank()) {
931+ addFilterInternal(rule.field, rule.operator, rule.value)
932+ }
933+ }
934+ }
935+ }
936+
937+ fun deleteSavedFilter(filterId: String) {
938+ viewModelScope.launch {
939+ savedFiltersMutex.withLock {
940+ when (repository.deleteFilter(filterId)) {
941+ is ApiResult.Success -> {
942+ _state.update {
943+ it.copy(
944+ savedFilters = it.savedFilters.copy(
945+ filters = it.savedFilters.filters.filter { f -> f.filterId != filterId },
946+ ),
947+ )
948+ }
949+ }
950+ is ApiResult.Error -> { /* Silently ignore — will show on next reload */ }
951+ }
952+ }
953+ }
954+ }
955+
474956 override fun onCleared() {
475957 super.onCleared()
476958 schemaJob?.cancel()