@@ -3,7 +3,6 @@ package com.parseable.android.ui.screens.logviewer
3import androidx.lifecycle.ViewModel
3import androidx.lifecycle.ViewModel
4import androidx.lifecycle.viewModelScope
4import androidx.lifecycle.viewModelScope
5import com.parseable.android.data.escapeIdentifier
5import com.parseable.android.data.escapeIdentifier
6−import com.parseable.android.data.escapeLikePattern
7import com.parseable.android.data.escapeSql
6import com.parseable.android.data.escapeSql
8import com.parseable.android.data.model.*
7import com.parseable.android.data.model.*
9import com.parseable.android.data.repository.ParseableRepository
8import com.parseable.android.data.repository.ParseableRepository
@@ -16,13 +15,8 @@ import kotlinx.coroutines.flow.asStateFlow
16import kotlinx.coroutines.flow.update
15import kotlinx.coroutines.flow.update
17import kotlinx.coroutines.isActive
16import kotlinx.coroutines.isActive
18import kotlinx.coroutines.launch
17import kotlinx.coroutines.launch
19−import kotlinx.coroutines.sync.Mutex
20−import kotlinx.coroutines.sync.withLock
21import kotlinx.serialization.json.JsonObject
18import kotlinx.serialization.json.JsonObject
22import kotlinx.serialization.json.jsonPrimitive
19import kotlinx.serialization.json.jsonPrimitive
23−import java.time.Instant
24−import java.time.LocalDateTime
25−import java.time.OffsetDateTime
26import java.time.ZoneOffset
20import java.time.ZoneOffset
27import java.time.ZonedDateTime
21import java.time.ZonedDateTime
28import java.time.format.DateTimeFormatter
22import java.time.format.DateTimeFormatter
@@ -40,21 +34,10 @@ enum class TimeRange(val label: String, val minutes: Long) {
40 CUSTOM("Custom", 0),
34 CUSTOM("Custom", 0),
43−/** The original (column, operator, value) a filter chip was built from. */
44−data class FilterCondition(
46− val operator: String,
50data class FilterState(
37data class FilterState(
51 val searchQuery: String = "",
38 val searchQuery: String = "",
52 val activeFilters: List<String> = emptyList(),
39 val activeFilters: List<String> = emptyList(),
53 val filterClauses: List<String> = emptyList(),
40 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(),
58 val customSql: String = "",
41 val customSql: String = "",
59 val isSearching: Boolean = false,
42 val isSearching: Boolean = false,
@@ -76,9 +59,7 @@ data class SavedFiltersState(
76data class LogViewerState(
59data class LogViewerState(
77 val streamName: String = "",
60 val streamName: String = "",
78 val logs: List<JsonObject> = emptyList(),
61 val logs: List<JsonObject> = emptyList(),
79− val logKeys: List<String> = emptyList(),
80 val columns: List<String> = emptyList(),
62 val columns: List<String> = emptyList(),
81− val searchableColumns: List<String> = emptyList(),
82 val isLoading: Boolean = false,
63 val isLoading: Boolean = false,
83 val error: String? = null,
64 val error: String? = null,
84 val selectedTimeRange: TimeRange = TimeRange.LAST_1H,
65 val selectedTimeRange: TimeRange = TimeRange.LAST_1H,
@@ -89,10 +70,6 @@ data class LogViewerState(
89 val hasMore: Boolean = false,
70 val hasMore: Boolean = false,
90 val customStartTime: Long? = null,
71 val customStartTime: Long? = null,
91 val customEndTime: Long? = null,
72 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,
97 // Convenience accessors for backward compatibility with Screen
74 // Convenience accessors for backward compatibility with Screen
98 val searchQuery: String get() = filters.searchQuery
75 val searchQuery: String get() = filters.searchQuery
@@ -120,53 +97,17 @@ class LogViewerViewModel @Inject constructor(
120 @Volatile private var streamingGeneration: Int = 0
97 @Volatile private var streamingGeneration: Int = 0
121 @Volatile private var lastSeenTimestamp: String? = null
98 @Volatile private var lastSeenTimestamp: String? = null
122 @Volatile private var consecutiveStreamingErrors: Int = 0
99 @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
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>()
135 private var searchJob: Job? = null
100 private var searchJob: Job? = null
136 private var schemaJob: Job? = null
101 private var schemaJob: Job? = null
137− private var refreshJob: Job? = null
139 companion object {
103 companion object {
140 private const val STREAMING_BASE_INTERVAL_MS = 3000L
104 private const val STREAMING_BASE_INTERVAL_MS = 3000L
141 private const val STREAMING_MAX_INTERVAL_MS = 30000L
105 private const val STREAMING_MAX_INTERVAL_MS = 30000L
142 private const val STREAMING_MAX_LOGS = 1000
106 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
146 private const val MAX_LOAD_LIMIT = 5000
107 private const val MAX_LOAD_LIMIT = 5000
147 private const val LOAD_MORE_INCREMENT = 500
108 private const val LOAD_MORE_INCREMENT = 500
148 private const val MAX_STREAMING_ERRORS = 5
109 private const val MAX_STREAMING_ERRORS = 5
149 private val ALLOWED_OPERATORS = setOf("=", "!=", "LIKE", "ILIKE", ">", "<", ">=", "<=", "IS NULL", "IS NOT NULL")
110 private val ALLOWED_OPERATORS = setOf("=", "!=", "LIKE", "ILIKE", ">", "<", ">=", "<=", "IS NULL", "IS NOT NULL")
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).
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"
172 fun initialize(streamName: String) {
113 fun initialize(streamName: String) {
@@ -182,9 +123,8 @@ class LogViewerViewModel @Inject constructor(
182 schemaJob = viewModelScope.launch {
123 schemaJob = viewModelScope.launch {
183 when (val result = repository.getStreamSchema(streamName)) {
124 when (val result = repository.getStreamSchema(streamName)) {
184 is ApiResult.Success -> {
125 is ApiResult.Success -> {
185− val columns = result.data.fields.map { it.name }.sorted()
126+ val columns = result.data.fields.map { it.name }
186− val searchable = columns.filter { !it.startsWith("p_") }
187− _state.update { it.copy(columns = columns, searchableColumns = searchable) }
127+ _state.update { it.copy(columns = columns) }
189 is ApiResult.Error -> { /* Schema loading failure is non-fatal */ }
129 is ApiResult.Error -> { /* Schema loading failure is non-fatal */ }
@@ -218,11 +158,7 @@ class LogViewerViewModel @Inject constructor(
220 fun onSearchQueryChange(query: String) {
160 fun onSearchQueryChange(query: String) {
221− // A new query invalidates the live-tail baseline and the paged-up limit, matching
222− // the time-range / custom-SQL handlers.
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())) }
161+ _state.update { it.copy(filters = it.filters.copy(searchQuery = query, isSearching = query.isNotBlank())) }
226 searchJob?.cancel()
162 searchJob?.cancel()
227 searchJob = viewModelScope.launch {
163 searchJob = viewModelScope.launch {
228 delay(300)
164 delay(300)
@@ -231,55 +167,28 @@ class LogViewerViewModel @Inject constructor(
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.
238− fun submitSearch() {
240− _state.update { it.copy(filters = it.filters.copy(isSearching = false)) }
245 fun addFilter(column: String, operator: String, value: String) {
170 fun addFilter(column: String, operator: String, value: String) {
246− addFilterInternal(column, operator, value)
250− private fun addFilterInternal(column: String, operator: String, value: String) {
251 if (operator !in ALLOWED_OPERATORS) return
171 if (operator !in ALLOWED_OPERATORS) return
252 val safeColumn = escapeIdentifier(column)
172 val safeColumn = escapeIdentifier(column)
173+ val safeValue = escapeSql(value)
253 val clause = when (operator) {
174 val clause = when (operator) {
254 "IS NULL", "IS NOT NULL" -> "\"$safeColumn\" $operator"
175 "IS NULL", "IS NOT NULL" -> "\"$safeColumn\" $operator"
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 '\\'"
176+ "LIKE", "ILIKE" -> "\"$safeColumn\" $operator '%$safeValue%'"
258− else -> "\"$safeColumn\" $operator '${escapeSql(value)}'"
177+ else -> "\"$safeColumn\" $operator '$safeValue'"
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" -> ""
266 val display = when (operator) {
179 val display = when (operator) {
267 "IS NULL", "IS NOT NULL" -> "$column $operator"
180 "IS NULL", "IS NOT NULL" -> "$column $operator"
268 else -> "$column $operator $value"
181 else -> "$column $operator $value"
270− // A new filter invalidates the live-tail baseline and the paged-up limit.
272 _state.update {
183 _state.update {
275 filters = it.filters.copy(
185 filters = it.filters.copy(
276 filterClauses = it.filters.filterClauses + clause,
186 filterClauses = it.filters.filterClauses + clause,
277 activeFilters = it.filters.activeFilters + display,
187 activeFilters = it.filters.activeFilters + display,
278− filterConditions = it.filters.filterConditions + FilterCondition(column, operator, conditionValue),
285 fun removeFilter(display: String) {
194 fun removeFilter(display: String) {
@@ -289,93 +198,81 @@ class LogViewerViewModel @Inject constructor(
289 if (index >= 0 && index < it.filters.filterClauses.size) {
198 if (index >= 0 && index < it.filters.filterClauses.size) {
290 removed = true
199 removed = true
293 filters = it.filters.copy(
201 filters = it.filters.copy(
294 activeFilters = it.filters.activeFilters.filterIndexed { i, _ -> i != index },
202 activeFilters = it.filters.activeFilters.filterIndexed { i, _ -> i != index },
295 filterClauses = it.filters.filterClauses.filterIndexed { i, _ -> i != index },
203 filterClauses = it.filters.filterClauses.filterIndexed { i, _ -> i != index },
296− filterConditions = it.filters.filterConditions.filterIndexed { i, _ -> i != index },
304− if (removed) {
210+ if (removed) refresh()
305− // A changed filter set invalidates the live-tail baseline.
311 fun clearFilters() {
213 fun clearFilters() {
313 _state.update {
214 _state.update {
314− it.copy(currentLimit = 500, filters = it.filters.copy(activeFilters = emptyList(), filterClauses = emptyList(), filterConditions = emptyList(), customSql = "", searchQuery = "", isSearching = false))
215+ it.copy(filters = it.filters.copy(activeFilters = emptyList(), filterClauses = emptyList()))
316 refresh()
217 refresh()
319 fun executeCustomSql(sql: String) {
220 fun executeCustomSql(sql: String) {
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().)
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()
221+ val trimmed = sql.trim()
327 if (!trimmed.uppercase().startsWith("SELECT")) {
222 if (!trimmed.uppercase().startsWith("SELECT")) {
328 _state.update { it.copy(error = "Only SELECT queries are allowed") }
223 _state.update { it.copy(error = "Only SELECT queries are allowed") }
331− // Enforce a LIMIT to prevent OOM from unbounded queries. Only honor a *top-level*
226+ // Enforce a LIMIT to prevent OOM from unbounded queries
332− // `LIMIT <n>` (optionally with OFFSET) anchored at the end of the statement — a
227+ val safeSql = if (!trimmed.uppercase().contains("LIMIT")) {
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) {
339 "$trimmed LIMIT $MAX_LOAD_LIMIT"
228 "$trimmed LIMIT $MAX_LOAD_LIMIT"
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.
346 _state.update { it.copy(filters = it.filters.copy(customSql = safeSql), currentLimit = 500) }
232 _state.update { it.copy(filters = it.filters.copy(customSql = safeSql), currentLimit = 500) }
347− refresh()
233+ viewModelScope.launch {
234+ _state.update { it.copy(isLoading = true, error = null) }
236+ val (startTime, endTime) = getTimeRange()
237+ when (val result = repository.queryLogsRaw(safeSql, startTime, endTime)) {
238+ is ApiResult.Success -> {
243+ hasMore = result.data.size >= it.currentLimit,
247+ is ApiResult.Error -> {
249+ it.copy(isLoading = false, error = result.userMessage)
350− private fun buildSearchClause(searchableColumns: List<String>, searchQuery: String): String? {
256+ private fun buildSearchClause(columns: List<String>, searchQuery: String): String? {
351 if (searchQuery.isBlank()) return null
257 if (searchQuery.isBlank()) return null
258+ val safeSearch = escapeSql(searchQuery)
259+ // Search across ALL non-internal columns (not just 5)
260+ val searchFields = columns.filter { !it.startsWith("p_") }
352− if (searchableColumns.isEmpty()) return null
261+ if (searchFields.isEmpty()) return null
353− val safeSearch = escapeLikePattern(searchQuery)
354− return searchableColumns.joinToString(" OR ") {
262+ return searchFields.joinToString(" OR ") {
355− "CAST(\"${escapeIdentifier(it)}\" AS VARCHAR) ILIKE '%$safeSearch%' ESCAPE '\\'"
263+ "\"${escapeIdentifier(it)}\" ILIKE '%$safeSearch%'"
359 fun refresh() {
267 fun refresh() {
360 if (_state.value.streamName.isEmpty()) return
268 if (_state.value.streamName.isEmpty()) return
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 {
270+ 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
373 _state.update {
271 _state.update {
375 isLoading = true,
273 isLoading = true,
376 error = null,
274 error = null,
377 streaming = it.streaming.copy(streamingError = null),
275 streaming = it.streaming.copy(streamingError = null),
378− pageAnchorTime = anchor,
@@ -383,24 +280,10 @@ class LogViewerViewModel @Inject constructor(
383 val current = _state.value
280 val current = _state.value
384 val (startTime, endTime) = getTimeRange()
281 val (startTime, endTime) = getTimeRange()
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) }
393− is ApiResult.Error -> {
394− _state.update { it.copy(isLoading = false, error = result.userMessage) }
400 // Build WHERE clause from filters + search
283 // Build WHERE clause from filters + search
401 val clauses = current.filters.filterClauses.toMutableList()
284 val clauses = current.filters.filterClauses.toMutableList()
402 if (current.filters.searchQuery.isNotBlank()) {
285 if (current.filters.searchQuery.isNotBlank()) {
403− val searchClause = buildSearchClause(current.searchableColumns, current.filters.searchQuery)
286+ val searchClause = buildSearchClause(current.columns, current.filters.searchQuery)
404 if (searchClause != null) {
287 if (searchClause != null) {
405 clauses.add("($searchClause)")
288 clauses.add("($searchClause)")
@@ -413,26 +296,20 @@ class LogViewerViewModel @Inject constructor(
414 val filterSql = clauses.joinToString(" AND ")
297 val filterSql = clauses.joinToString(" AND ")
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
420 when (val result = repository.queryLogs(
299 when (val result = repository.queryLogs(
421 stream = current.streamName,
300 stream = current.streamName,
422 startTime = startTime,
301 startTime = startTime,
423 endTime = endTime,
302 endTime = endTime,
424 filterSql = filterSql,
303 filterSql = filterSql,
425− limit = requestLimit,
304+ limit = current.currentLimit,
427 is ApiResult.Success -> {
306 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)
434 _state.update {
307 _state.update {
435− it.copy(logs = logs, logKeys = computeLogKeys(logs), isLoading = false, hasMore = hasMore)
308+ it.copy(
311+ hasMore = result.data.size >= it.currentLimit,
438 is ApiResult.Error -> {
315 is ApiResult.Error -> {
@@ -444,29 +321,8 @@ class LogViewerViewModel @Inject constructor(
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.
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) }
465 fun loadMore() {
324 fun loadMore() {
466 val current = _state.value
325 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
470 if (current.currentLimit >= MAX_LOAD_LIMIT) {
326 if (current.currentLimit >= MAX_LOAD_LIMIT) {
471 _state.update { it.copy(hasMore = false) }
327 _state.update { it.copy(hasMore = false) }
@@ -489,9 +345,7 @@ class LogViewerViewModel @Inject constructor(
490 return Pair(start.format(dateFormatter), end.format(dateFormatter))
346 return Pair(start.format(dateFormatter), end.format(dateFormatter))
492− val now = current.pageAnchorTime
493− ?.let { ZonedDateTime.ofInstant(java.time.Instant.ofEpochMilli(it), ZoneOffset.UTC) }
494− ?: ZonedDateTime.now(ZoneOffset.UTC)
348+ val now = ZonedDateTime.now(ZoneOffset.UTC)
495 val start = now.minusMinutes(current.selectedTimeRange.minutes)
349 val start = now.minusMinutes(current.selectedTimeRange.minutes)
496 return Pair(
350 return Pair(
497 start.format(dateFormatter),
351 start.format(dateFormatter),
@@ -499,17 +353,6 @@ class LogViewerViewModel @Inject constructor(
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.
506− fun setViewingTop(atTop: Boolean) {
508− if (atTop && _state.value.streaming.streamingNewCount != 0) {
509− _state.update { it.copy(streaming = it.streaming.copy(streamingNewCount = 0)) }
513 fun toggleStreaming() {
356 fun toggleStreaming() {
514 if (_state.value.streaming.isStreaming) {
357 if (_state.value.streaming.isStreaming) {
515 stopStreaming()
358 stopStreaming()
@@ -519,10 +362,6 @@ class LogViewerViewModel @Inject constructor(
521 private fun startStreaming() {
364 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
526 stopStreaming()
365 stopStreaming()
527 consecutiveStreamingErrors = 0
366 consecutiveStreamingErrors = 0
528 val generation = ++streamingGeneration
367 val generation = ++streamingGeneration
@@ -530,12 +369,6 @@ class LogViewerViewModel @Inject constructor(
530 it.copy(streaming = StreamingState(isStreaming = true, streamingNewCount = 0))
369 it.copy(streaming = StreamingState(isStreaming = true, streamingNewCount = 0))
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()
539 // Set the baseline timestamp to "now" so we only poll for new logs
372 // Set the baseline timestamp to "now" so we only poll for new logs
540 lastSeenTimestamp = ZonedDateTime.now(ZoneOffset.UTC).format(dateFormatter)
373 lastSeenTimestamp = ZonedDateTime.now(ZoneOffset.UTC).format(dateFormatter)
@@ -558,13 +391,6 @@ class LogViewerViewModel @Inject constructor(
560 fun stopStreaming() {
393 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
568 streamingGeneration++
394 streamingGeneration++
569 streamingJob?.cancel()
395 streamingJob?.cancel()
570 streamingJob = null
396 streamingJob = null
@@ -579,49 +405,6 @@ class LogViewerViewModel @Inject constructor(
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.
587− private fun logIdentity(log: JsonObject): String = log.toString()
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.
595− private fun normalizeBoundaryTimestamp(raw: String): String? {
596− val isoLike = if (' ' in raw && 'T' !in raw) raw.replaceFirst(' ', 'T') else raw
598− Instant.parse(isoLike)
599− } catch (_: Exception) {
601− OffsetDateTime.parse(isoLike).toInstant()
602− } catch (_: Exception) {
604− LocalDateTime.parse(isoLike).toInstant(ZoneOffset.UTC)
605− } catch (_: Exception) {
610− return instant.atOffset(ZoneOffset.UTC).format(dateFormatter)
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()) {
625 private suspend fun pollNewLogs() {
408 private suspend fun pollNewLogs() {
626 val current = _state.value
409 val current = _state.value
627 if (current.streamName.isEmpty()) return
410 if (current.streamName.isEmpty()) return
@@ -632,118 +415,46 @@ class LogViewerViewModel @Inject constructor(
632 // Build WHERE clause from active filters + search
415 // Build WHERE clause from active filters + search
633 val clauses = current.filters.filterClauses.toMutableList()
416 val clauses = current.filters.filterClauses.toMutableList()
634 if (current.filters.searchQuery.isNotBlank()) {
417 if (current.filters.searchQuery.isNotBlank()) {
635− val searchClause = buildSearchClause(current.searchableColumns, current.filters.searchQuery)
418+ val searchClause = buildSearchClause(current.columns, current.filters.searchQuery)
636 if (searchClause != null) {
419 if (searchClause != null) {
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)")
420+ clauses.add(searchClause)
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.
647− streaming = it.streaming.copy(
648− streamingError = "Search unavailable: stream schema not loaded",
656 val whereClause = if (clauses.isNotEmpty()) " WHERE ${clauses.joinToString(" AND ")}" else ""
424 val whereClause = if (clauses.isNotEmpty()) " WHERE ${clauses.joinToString(" AND ")}" else ""
657 val safeName = escapeIdentifier(current.streamName)
425 val safeName = escapeIdentifier(current.streamName)
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"
426+ val sql = "SELECT * FROM \"$safeName\"$whereClause ORDER BY p_timestamp DESC LIMIT 200"
664 when (val result = repository.queryLogsRaw(sql, startTime, endTime)) {
428 when (val result = repository.queryLogsRaw(sql, startTime, endTime)) {
665 is ApiResult.Success -> {
429 is ApiResult.Success -> {
666 val newLogs = result.data
430 val newLogs = result.data
667 if (newLogs.isNotEmpty()) {
431 if (newLogs.isNotEmpty()) {
668− // Advance the last-seen boundary. Rows are ordered newest-first, so
432+ // Update the last seen timestamp to the most recent log
669− // normally we jump to the newest. But a full page means the LIMIT may
433+ val newestTimestamp = try {
670− // have capped off matching rows OLDER than the newest in this batch
434+ newLogs.firstOrNull()
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
435+ ?.get("p_timestamp")?.jsonPrimitive?.content
679 } catch (_: Exception) {
436 } catch (_: Exception) {
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 {
439+ if (newestTimestamp != null) {
697− newLogs.firstOrNull()?.get("p_timestamp")?.jsonPrimitive?.content
698− } catch (_: Exception) {
701− val newestNormalized = newestTimestamp?.let { normalizeBoundaryTimestamp(it) }
702− if (newestNormalized != null && newestNormalized != lastSeenTimestamp) {
703− normalized = newestNormalized
706− if (normalized != null) {
707− lastSeenTimestamp = normalized
440+ lastSeenTimestamp = newestTimestamp
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()
719− if (freshLogs.isEmpty()) {
720− _state.update { it.copy(streaming = it.streaming.copy(streamingError = null)) }
722− _state.update { state ->
443+ _state.update { state ->
723− // Prepend new logs, cap total — avoid full intermediate list allocation
444+ // Prepend new logs, cap total
724− val remaining = (STREAMING_MAX_LOGS - freshLogs.size).coerceAtLeast(0)
445+ val combined = newLogs + state.logs
725− val capped = ArrayList<JsonObject>(freshLogs.size + remaining).apply {
446+ val capped = if (combined.size > STREAMING_MAX_LOGS) {
726− addAll(freshLogs)
447+ combined.take(STREAMING_MAX_LOGS)
727− val oldLogs = state.logs
728− addAll(oldLogs.subList(0, remaining.coerceAtMost(oldLogs.size)))
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) {
740− } else {
448+ } else {
741− state.streaming.streamingNewCount + freshLogs.size
449+ combined
743− streamingError = null,
453+ streaming = state.streaming.copy(
454+ streamingNewCount = state.streaming.streamingNewCount + newLogs.size,
455+ streamingError = null,
749 consecutiveStreamingErrors = 0
460 consecutiveStreamingErrors = 0
@@ -770,37 +481,19 @@ class LogViewerViewModel @Inject constructor(
771 // --- Saved Filters (synced with Parseable server) ---
482 // --- Saved Filters (synced with Parseable server) ---
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()
779 fun loadSavedFilters() {
484 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 {
485 viewModelScope.launch {
785− savedFiltersMutex.withLock {
786− _state.update { it.copy(savedFilters = it.savedFilters.copy(isLoading = true, error = null)) }
486+ _state.update { it.copy(savedFilters = it.savedFilters.copy(isLoading = true, error = null)) }
787− when (val result = repository.listFilters()) {
487+ when (val result = repository.listFilters()) {
788− is ApiResult.Success -> {
488+ is ApiResult.Success -> {
789− if (_state.value.streamName != stream) {
790− _state.update { it.copy(savedFilters = it.savedFilters.copy(isLoading = false)) }
793− val streamFilters = result.data
489+ val streamFilters = result.data.filter { it.streamName == _state.value.streamName }
794− .filter { it.streamName == stream }
795− .sortedBy { it.filterName.lowercase() }
796− _state.update {
490+ _state.update {
797− it.copy(savedFilters = it.savedFilters.copy(filters = streamFilters, isLoading = false))
491+ it.copy(savedFilters = it.savedFilters.copy(filters = streamFilters, isLoading = false))
800− is ApiResult.Error -> {
494+ is ApiResult.Error -> {
801− _state.update {
495+ _state.update {
802− it.copy(savedFilters = it.savedFilters.copy(isLoading = false, error = result.userMessage))
496+ it.copy(savedFilters = it.savedFilters.copy(isLoading = false, error = result.userMessage))
@@ -812,7 +505,6 @@ class LogViewerViewModel @Inject constructor(
812 if (current.streamName.isEmpty()) return
505 if (current.streamName.isEmpty()) return
814 viewModelScope.launch {
507 viewModelScope.launch {
815− savedFiltersMutex.withLock {
816 _state.update { it.copy(savedFilters = it.savedFilters.copy(isSaving = true, error = null)) }
508 _state.update { it.copy(savedFilters = it.savedFilters.copy(isSaving = true, error = null)) }
818 val query = if (current.filters.customSql.isNotBlank()) {
510 val query = if (current.filters.customSql.isNotBlank()) {
@@ -821,13 +513,15 @@ class LogViewerViewModel @Inject constructor(
821 filterQuery = current.filters.customSql,
513 filterQuery = current.filters.customSql,
824− val ruleGroups = current.filters.filterConditions
516+ val ruleGroups = current.filters.activeFilters.zip(current.filters.filterClauses)
825− .mapIndexed { index, condition ->
517+ .mapIndexed { index, (display, _) ->
518+ // Parse display back to field/operator/value
519+ val parts = parseFilterDisplay(display)
826 FilterRule(
520 FilterRule(
827 id = "rule_$index",
521 id = "rule_$index",
828− field = condition.column,
522+ field = parts.first,
829− value = condition.value,
523+ value = parts.third,
830− operator = condition.operator,
524+ operator = parts.second,
833 val builder = if (ruleGroups.isNotEmpty()) {
527 val builder = if (ruleGroups.isNotEmpty()) {
@@ -874,7 +568,6 @@ class LogViewerViewModel @Inject constructor(
@@ -884,15 +577,9 @@ class LogViewerViewModel @Inject constructor(
884 when (filter.query.filterType) {
577 when (filter.query.filterType) {
885 "sql" -> {
578 "sql" -> {
886 val sql = filter.query.filterQuery ?: return
579 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 {
580 _state.update {
895− filters = FilterState(),
582+ filters = FilterState(customSql = sql),
896 currentLimit = 500,
583 currentLimit = 500,
@@ -928,7 +615,7 @@ class LogViewerViewModel @Inject constructor(
928 for (group in builder.rules) {
615 for (group in builder.rules) {
929 for (rule in group.rules) {
616 for (rule in group.rules) {
930 if (rule.field.isNotBlank()) {
617 if (rule.field.isNotBlank()) {
931− addFilterInternal(rule.field, rule.operator, rule.value)
618+ addFilter(rule.field, rule.operator, rule.value)
@@ -936,21 +623,32 @@ class LogViewerViewModel @Inject constructor(
937 fun deleteSavedFilter(filterId: String) {
624 fun deleteSavedFilter(filterId: String) {
938 viewModelScope.launch {
625 viewModelScope.launch {
939− savedFiltersMutex.withLock {
940− when (repository.deleteFilter(filterId)) {
626+ when (repository.deleteFilter(filterId)) {
941− is ApiResult.Success -> {
627+ is ApiResult.Success -> {
942− _state.update {
628+ _state.update {
943− it.copy(
629+ it.copy(
944− savedFilters = it.savedFilters.copy(
630+ savedFilters = it.savedFilters.copy(
945− filters = it.savedFilters.filters.filter { f -> f.filterId != filterId },
631+ filters = it.savedFilters.filters.filter { f -> f.filterId != filterId },
950− is ApiResult.Error -> { /* Silently ignore — will show on next reload */ }
636+ is ApiResult.Error -> { /* Silently ignore — will show on next reload */ }
641+ /** Parse a display string like "column = value" back into (field, operator, value) */
642+ private fun parseFilterDisplay(display: String): Triple<String, String, String> {
643+ for (op in listOf("IS NOT NULL", "IS NULL", "ILIKE", "LIKE", "!=", ">=", "<=", "=", ">", "<")) {
644+ val idx = display.indexOf(" $op")
646+ val field = display.substring(0, idx)
647+ val value = display.substring(idx + op.length + 1).trim()
648+ return Triple(field, op, value)
651+ return Triple(display, "=", "")
956 override fun onCleared() {
654 override fun onCleared() {