git.iko.soy/parseable-android
on 3b7ea33 help sign in
parseable-android/app/src/main/java/com/parseable/android/ui/screens/logviewer/LogViewerViewModel.kt
687 lines 26 kB history raw
1package com.parseable.android.ui.screens.logviewer
3import androidx.lifecycle.ViewModel
4import androidx.lifecycle.viewModelScope
5import com.parseable.android.data.escapeIdentifier
6import com.parseable.android.data.escapeSql
7import com.parseable.android.data.model.*
8import com.parseable.android.data.repository.ParseableRepository
9import dagger.hilt.android.lifecycle.HiltViewModel
10import kotlinx.coroutines.Job
11import kotlinx.coroutines.delay
12import kotlinx.coroutines.flow.MutableStateFlow
13import kotlinx.coroutines.flow.StateFlow
14import kotlinx.coroutines.flow.asStateFlow
15import kotlinx.coroutines.flow.update
16import kotlinx.coroutines.isActive
17import kotlinx.coroutines.launch
18import kotlinx.serialization.json.JsonObject
19import kotlinx.serialization.json.jsonPrimitive
20import java.time.ZoneOffset
21import java.time.ZonedDateTime
22import java.time.format.DateTimeFormatter
23import javax.inject.Inject
25enum class TimeRange(val label: String, val minutes: Long) {
26 LAST_5M("5m", 5),
27 LAST_15M("15m", 15),
28 LAST_30M("30m", 30),
29 LAST_1H("1h", 60),
30 LAST_6H("6h", 360),
31 LAST_24H("24h", 1440),
32 LAST_7D("7d", 10080),
33 LAST_30D("30d", 43200),
34 CUSTOM("Custom", 0),
37data class FilterState(
38 val searchQuery: String = "",
39 val activeFilters: List<String> = emptyList(),
40 val filterClauses: List<String> = emptyList(),
41 val customSql: String = "",
42 val isSearching: Boolean = false,
45data class StreamingState(
46 val isStreaming: Boolean = false,
47 val streamingNewCount: Int = 0,
48 val streamingError: String? = null,
49 val currentIntervalMs: Long = 3000L,
52data class SavedFiltersState(
53 val filters: List<SavedFilter> = emptyList(),
54 val isLoading: Boolean = false,
55 val isSaving: Boolean = false,
56 val error: String? = null,
59data class LogViewerState(
60 val streamName: String = "",
61 val logs: List<JsonObject> = emptyList(),
62 val logKeys: List<String> = emptyList(),
63 val columns: List<String> = emptyList(),
64 val searchableColumns: List<String> = emptyList(),
65 val isLoading: Boolean = false,
66 val error: String? = null,
67 val selectedTimeRange: TimeRange = TimeRange.LAST_1H,
68 val filters: FilterState = FilterState(),
69 val streaming: StreamingState = StreamingState(),
70 val savedFilters: SavedFiltersState = SavedFiltersState(),
71 val currentLimit: Int = 500,
72 val hasMore: Boolean = false,
73 val customStartTime: Long? = null,
74 val customEndTime: Long? = null,
75) {
76 // Convenience accessors for backward compatibility with Screen
77 val searchQuery: String get() = filters.searchQuery
78 val activeFilters: List<String> get() = filters.activeFilters
79 val filterClauses: List<String> get() = filters.filterClauses
80 val customSql: String get() = filters.customSql
81 val isStreaming: Boolean get() = streaming.isStreaming
82 val streamingNewCount: Int get() = streaming.streamingNewCount
83 val streamingError: String? get() = streaming.streamingError
84 val currentIntervalMs: Long get() = streaming.currentIntervalMs
85 val isSearching: Boolean get() = filters.isSearching
88@HiltViewModel
89class LogViewerViewModel @Inject constructor(
90 private val repository: ParseableRepository,
91) : ViewModel() {
93 private val _state = MutableStateFlow(LogViewerState())
94 val state: StateFlow<LogViewerState> = _state.asStateFlow()
96 private val dateFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSSSS'+00:00'")
98 private var streamingJob: Job? = null
99 @Volatile private var streamingGeneration: Int = 0
100 @Volatile private var lastSeenTimestamp: String? = null
101 @Volatile private var consecutiveStreamingErrors: Int = 0
102 private var searchJob: Job? = null
103 private var schemaJob: Job? = null
105 companion object {
106 private const val STREAMING_BASE_INTERVAL_MS = 3000L
107 private const val STREAMING_MAX_INTERVAL_MS = 30000L
108 private const val STREAMING_MAX_LOGS = 1000
109 private const val MAX_LOAD_LIMIT = 5000
110 private const val LOAD_MORE_INCREMENT = 500
111 private const val MAX_STREAMING_ERRORS = 5
112 private val ALLOWED_OPERATORS = setOf("=", "!=", "LIKE", "ILIKE", ">", "<", ">=", "<=", "IS NULL", "IS NOT NULL")
114 /**
115 * Produces deterministic keys for a list of log entries so that expanded state
116 * survives list mutations (e.g., new logs prepended during streaming).
117 */
118 internal fun computeLogKeys(logs: List<JsonObject>): List<String> {
119 val seen = mutableMapOf<String, Int>()
120 return logs.map { log ->
121 val ts = log["p_timestamp"]?.toString()
122 val meta = log["p_metadata"]?.toString()
123 val tag = log["p_tags"]?.toString()
124 val base = if (ts != null) "$ts|${meta.orEmpty()}|${tag.orEmpty()}" else log.hashCode().toString()
125 val count = seen.getOrDefault(base, 0)
126 seen[base] = count + 1
127 if (count == 0) base else "$base|$count"
128 }
129 }
130 }
132 fun initialize(streamName: String) {
133 if (_state.value.streamName == streamName) return
134 stopStreaming()
135 _state.update { it.copy(streamName = streamName) }
136 loadSchema(streamName)
137 refresh()
138 }
140 private fun loadSchema(streamName: String) {
141 schemaJob?.cancel()
142 schemaJob = viewModelScope.launch {
143 when (val result = repository.getStreamSchema(streamName)) {
144 is ApiResult.Success -> {
145 val columns = result.data.fields.map { it.name }.sorted()
146 val searchable = columns.filter { !it.startsWith("p_") }
147 _state.update { it.copy(columns = columns, searchableColumns = searchable) }
148 }
149 is ApiResult.Error -> { /* Schema loading failure is non-fatal */ }
150 }
151 }
152 }
154 fun onTimeRangeChange(range: TimeRange) {
155 stopStreaming()
156 _state.update {
157 it.copy(
158 selectedTimeRange = range,
159 currentLimit = 500,
160 customStartTime = if (range != TimeRange.CUSTOM) null else it.customStartTime,
161 customEndTime = if (range != TimeRange.CUSTOM) null else it.customEndTime,
162 )
163 }
164 if (range != TimeRange.CUSTOM) refresh()
165 }
167 fun setCustomTimeRange(startMillis: Long, endMillis: Long) {
168 stopStreaming()
169 _state.update {
170 it.copy(
171 selectedTimeRange = TimeRange.CUSTOM,
172 customStartTime = startMillis,
173 customEndTime = endMillis,
174 currentLimit = 500,
175 )
176 }
177 refresh()
178 }
180 fun onSearchQueryChange(query: String) {
181 _state.update { it.copy(filters = it.filters.copy(searchQuery = query, isSearching = query.isNotBlank())) }
182 searchJob?.cancel()
183 searchJob = viewModelScope.launch {
184 delay(300)
185 _state.update { it.copy(filters = it.filters.copy(isSearching = false)) }
186 refresh()
187 }
188 }
190 fun addFilter(column: String, operator: String, value: String) {
191 if (operator !in ALLOWED_OPERATORS) return
192 val safeColumn = escapeIdentifier(column)
193 val safeValue = escapeSql(value)
194 val clause = when (operator) {
195 "IS NULL", "IS NOT NULL" -> "\"$safeColumn\" $operator"
196 "LIKE", "ILIKE" -> "\"$safeColumn\" $operator '%$safeValue%'"
197 else -> "\"$safeColumn\" $operator '$safeValue'"
198 }
199 val display = when (operator) {
200 "IS NULL", "IS NOT NULL" -> "$column $operator"
201 else -> "$column $operator $value"
202 }
203 _state.update {
204 it.copy(
205 filters = it.filters.copy(
206 filterClauses = it.filters.filterClauses + clause,
207 activeFilters = it.filters.activeFilters + display,
208 ),
209 )
210 }
211 refresh()
212 }
214 fun removeFilter(display: String) {
215 var removed = false
216 _state.update {
217 val index = it.filters.activeFilters.indexOf(display)
218 if (index >= 0 && index < it.filters.filterClauses.size) {
219 removed = true
220 it.copy(
221 filters = it.filters.copy(
222 activeFilters = it.filters.activeFilters.filterIndexed { i, _ -> i != index },
223 filterClauses = it.filters.filterClauses.filterIndexed { i, _ -> i != index },
224 ),
225 )
226 } else {
227 it
228 }
229 }
230 if (removed) refresh()
231 }
233 fun clearFilters() {
234 _state.update {
235 it.copy(filters = it.filters.copy(activeFilters = emptyList(), filterClauses = emptyList()))
236 }
237 refresh()
238 }
240 fun executeCustomSql(sql: String) {
241 val trimmed = sql.trim()
242 if (!trimmed.uppercase().startsWith("SELECT")) {
243 _state.update { it.copy(error = "Only SELECT queries are allowed") }
244 return
245 }
246 // Enforce a LIMIT to prevent OOM from unbounded queries
247 val safeSql = if (!trimmed.uppercase().contains("LIMIT")) {
248 "$trimmed LIMIT $MAX_LOAD_LIMIT"
249 } else {
250 trimmed
251 }
252 _state.update { it.copy(filters = it.filters.copy(customSql = safeSql), currentLimit = 500) }
253 viewModelScope.launch {
254 _state.update { it.copy(isLoading = true, error = null) }
256 val (startTime, endTime) = getTimeRange()
257 when (val result = repository.queryLogsRaw(safeSql, startTime, endTime)) {
258 is ApiResult.Success -> {
259 val keys = computeLogKeys(result.data)
260 _state.update {
261 it.copy(
262 logs = result.data,
263 logKeys = keys,
264 isLoading = false,
265 hasMore = result.data.size >= it.currentLimit,
266 )
267 }
268 }
269 is ApiResult.Error -> {
270 _state.update {
271 it.copy(isLoading = false, error = result.userMessage)
272 }
273 }
274 }
275 }
276 }
278 private fun buildSearchClause(searchableColumns: List<String>, searchQuery: String): String? {
279 if (searchQuery.isBlank()) return null
280 if (searchableColumns.isEmpty()) return null
281 val safeSearch = escapeSql(searchQuery)
282 return searchableColumns.joinToString(" OR ") {
283 "\"${escapeIdentifier(it)}\" ILIKE '%$safeSearch%'"
284 }
285 }
287 fun refresh() {
288 if (_state.value.streamName.isEmpty()) return
290 viewModelScope.launch {
291 _state.update {
292 it.copy(
293 isLoading = true,
294 error = null,
295 streaming = it.streaming.copy(streamingError = null),
296 )
297 }
299 // Snapshot state inside the coroutine so we always see the latest filters/columns
300 val current = _state.value
301 val (startTime, endTime) = getTimeRange()
303 // Build WHERE clause from filters + search
304 val clauses = current.filters.filterClauses.toMutableList()
305 if (current.filters.searchQuery.isNotBlank()) {
306 val searchClause = buildSearchClause(current.searchableColumns, current.filters.searchQuery)
307 if (searchClause != null) {
308 clauses.add("($searchClause)")
309 } else {
310 _state.update {
311 it.copy(isLoading = false, error = "Search unavailable: stream schema not loaded")
312 }
313 return@launch
314 }
315 }
317 val filterSql = clauses.joinToString(" AND ")
319 when (val result = repository.queryLogs(
320 stream = current.streamName,
321 startTime = startTime,
322 endTime = endTime,
323 filterSql = filterSql,
324 limit = current.currentLimit,
325 )) {
326 is ApiResult.Success -> {
327 val keys = computeLogKeys(result.data)
328 _state.update {
329 it.copy(
330 logs = result.data,
331 logKeys = keys,
332 isLoading = false,
333 hasMore = result.data.size >= it.currentLimit,
334 )
335 }
336 }
337 is ApiResult.Error -> {
338 _state.update {
339 it.copy(isLoading = false, error = result.userMessage)
340 }
341 }
342 }
343 }
344 }
346 fun loadMore() {
347 val current = _state.value
348 if (current.currentLimit >= MAX_LOAD_LIMIT) {
349 _state.update { it.copy(hasMore = false) }
350 return
351 }
352 val newLimit = (current.currentLimit + LOAD_MORE_INCREMENT).coerceAtMost(MAX_LOAD_LIMIT)
353 _state.update { it.copy(currentLimit = newLimit) }
354 refresh()
355 }
357 private fun getTimeRange(): Pair<String, String> {
358 val current = _state.value
359 if (current.selectedTimeRange == TimeRange.CUSTOM &&
360 current.customStartTime != null && current.customEndTime != null
361 ) {
362 val start = ZonedDateTime.ofInstant(
363 java.time.Instant.ofEpochMilli(current.customStartTime), ZoneOffset.UTC
364 )
365 val end = ZonedDateTime.ofInstant(
366 java.time.Instant.ofEpochMilli(current.customEndTime), ZoneOffset.UTC
367 )
368 return Pair(start.format(dateFormatter), end.format(dateFormatter))
369 }
370 val now = ZonedDateTime.now(ZoneOffset.UTC)
371 val start = now.minusMinutes(current.selectedTimeRange.minutes)
372 return Pair(
373 start.format(dateFormatter),
374 now.format(dateFormatter),
375 )
376 }
378 fun toggleStreaming() {
379 if (_state.value.streaming.isStreaming) {
380 stopStreaming()
381 } else {
382 startStreaming()
383 }
384 }
386 private fun startStreaming() {
387 stopStreaming()
388 consecutiveStreamingErrors = 0
389 val generation = ++streamingGeneration
390 _state.update {
391 it.copy(streaming = StreamingState(isStreaming = true, streamingNewCount = 0))
392 }
394 // Set the baseline timestamp to "now" so we only poll for new logs
395 lastSeenTimestamp = ZonedDateTime.now(ZoneOffset.UTC).format(dateFormatter)
397 streamingJob = viewModelScope.launch {
398 while (isActive && streamingGeneration == generation) {
399 pollNewLogs()
400 // Exponential backoff on errors, normal interval on success
401 val intervalMs = if (consecutiveStreamingErrors > 0) {
402 (STREAMING_BASE_INTERVAL_MS * (1L shl consecutiveStreamingErrors.coerceAtMost(4)))
403 .coerceAtMost(STREAMING_MAX_INTERVAL_MS)
404 } else {
405 STREAMING_BASE_INTERVAL_MS
406 }
407 _state.update {
408 it.copy(streaming = it.streaming.copy(currentIntervalMs = intervalMs))
409 }
410 delay(intervalMs)
411 }
412 }
413 }
415 fun stopStreaming() {
416 streamingGeneration++
417 streamingJob?.cancel()
418 streamingJob = null
419 _state.update {
420 it.copy(streaming = it.streaming.copy(isStreaming = false))
421 }
422 }
424 fun dismissStreamingError() {
425 _state.update {
426 it.copy(streaming = it.streaming.copy(streamingError = null))
427 }
428 }
430 private suspend fun pollNewLogs() {
431 val current = _state.value
432 if (current.streamName.isEmpty()) return
434 val startTime = lastSeenTimestamp ?: return
435 val endTime = ZonedDateTime.now(ZoneOffset.UTC).format(dateFormatter)
437 // Build WHERE clause from active filters + search
438 val clauses = current.filters.filterClauses.toMutableList()
439 if (current.filters.searchQuery.isNotBlank()) {
440 val searchClause = buildSearchClause(current.searchableColumns, current.filters.searchQuery)
441 if (searchClause != null) {
442 clauses.add(searchClause)
443 }
444 }
446 val whereClause = if (clauses.isNotEmpty()) " WHERE ${clauses.joinToString(" AND ")}" else ""
447 val safeName = escapeIdentifier(current.streamName)
448 val sql = "SELECT * FROM \"$safeName\"$whereClause ORDER BY p_timestamp DESC LIMIT 200"
450 when (val result = repository.queryLogsRaw(sql, startTime, endTime)) {
451 is ApiResult.Success -> {
452 val newLogs = result.data
453 if (newLogs.isNotEmpty()) {
454 // Update the last seen timestamp to the most recent log
455 val newestTimestamp = try {
456 newLogs.firstOrNull()
457 ?.get("p_timestamp")?.jsonPrimitive?.content
458 } catch (_: Exception) {
459 null
460 }
461 if (newestTimestamp != null) {
462 lastSeenTimestamp = newestTimestamp
463 }
465 _state.update { state ->
466 // Prepend new logs, cap total — avoid full intermediate list allocation
467 val remaining = (STREAMING_MAX_LOGS - newLogs.size).coerceAtLeast(0)
468 val capped = ArrayList<JsonObject>(newLogs.size + remaining).apply {
469 addAll(newLogs)
470 val oldLogs = state.logs
471 addAll(oldLogs.subList(0, remaining.coerceAtMost(oldLogs.size)))
472 }
473 state.copy(
474 logs = capped,
475 logKeys = computeLogKeys(capped),
476 streaming = state.streaming.copy(
477 streamingNewCount = state.streaming.streamingNewCount + newLogs.size,
478 streamingError = null,
479 ),
480 )
481 }
482 }
483 consecutiveStreamingErrors = 0
484 }
485 is ApiResult.Error -> {
486 consecutiveStreamingErrors++
487 if (consecutiveStreamingErrors >= MAX_STREAMING_ERRORS) {
488 stopStreaming()
489 _state.update {
490 it.copy(error = "Streaming stopped after repeated errors: ${result.userMessage}")
491 }
492 } else if (consecutiveStreamingErrors >= 3) {
493 _state.update {
494 it.copy(
495 streaming = it.streaming.copy(
496 streamingError = "Streaming interrupted: ${result.userMessage}. Retrying..."
497 ),
498 )
499 }
500 }
501 }
502 }
503 }
505 // --- Saved Filters (synced with Parseable server) ---
507 fun loadSavedFilters() {
508 viewModelScope.launch {
509 _state.update { it.copy(savedFilters = it.savedFilters.copy(isLoading = true, error = null)) }
510 when (val result = repository.listFilters()) {
511 is ApiResult.Success -> {
512 val streamFilters = result.data
513 .filter { it.streamName == _state.value.streamName }
514 .sortedBy { it.filterName.lowercase() }
515 _state.update {
516 it.copy(savedFilters = it.savedFilters.copy(filters = streamFilters, isLoading = false))
517 }
518 }
519 is ApiResult.Error -> {
520 _state.update {
521 it.copy(savedFilters = it.savedFilters.copy(isLoading = false, error = result.userMessage))
522 }
523 }
524 }
525 }
526 }
528 fun saveCurrentFilter(name: String) {
529 val current = _state.value
530 if (current.streamName.isEmpty()) return
532 viewModelScope.launch {
533 _state.update { it.copy(savedFilters = it.savedFilters.copy(isSaving = true, error = null)) }
535 val query = if (current.filters.customSql.isNotBlank()) {
536 SavedFilterQuery(
537 filterType = "sql",
538 filterQuery = current.filters.customSql,
539 )
540 } else {
541 val ruleGroups = current.filters.activeFilters.zip(current.filters.filterClauses)
542 .mapIndexed { index, (display, _) ->
543 // Parse display back to field/operator/value
544 val parts = parseFilterDisplay(display)
545 FilterRule(
546 id = "rule_$index",
547 field = parts.first,
548 value = parts.third,
549 operator = parts.second,
550 )
551 }
552 val builder = if (ruleGroups.isNotEmpty()) {
553 FilterBuilder(
554 id = "root",
555 combinator = "and",
556 rules = listOf(
557 FilterRuleGroup(
558 id = "group_0",
559 combinator = "and",
560 rules = ruleGroups,
561 ),
562 ),
563 )
564 } else null
566 SavedFilterQuery(
567 filterType = if (current.filters.searchQuery.isNotBlank()) "search" else "filter",
568 filterQuery = current.filters.searchQuery.ifBlank { null },
569 filterBuilder = builder,
570 )
571 }
573 val filter = SavedFilter(
574 filterName = name,
575 streamName = current.streamName,
576 query = query,
577 )
579 when (val result = repository.createFilter(filter)) {
580 is ApiResult.Success -> {
581 _state.update {
582 it.copy(
583 savedFilters = it.savedFilters.copy(
584 isSaving = false,
585 filters = it.savedFilters.filters + result.data,
586 ),
587 )
588 }
589 }
590 is ApiResult.Error -> {
591 _state.update {
592 it.copy(savedFilters = it.savedFilters.copy(isSaving = false, error = result.userMessage))
593 }
594 }
595 }
596 }
597 }
599 fun applySavedFilter(filter: SavedFilter) {
600 stopStreaming()
602 when (filter.query.filterType) {
603 "sql" -> {
604 val sql = filter.query.filterQuery ?: return
605 _state.update {
606 it.copy(
607 filters = FilterState(customSql = sql),
608 currentLimit = 500,
609 )
610 }
611 executeCustomSql(sql)
612 }
613 "search" -> {
614 val searchQuery = filter.query.filterQuery ?: ""
615 _state.update {
616 it.copy(
617 filters = FilterState(searchQuery = searchQuery),
618 currentLimit = 500,
619 )
620 }
621 // Also apply builder rules if present
622 applyBuilderRules(filter.query.filterBuilder)
623 refresh()
624 }
625 else -> {
626 _state.update {
627 it.copy(
628 filters = FilterState(),
629 currentLimit = 500,
630 )
631 }
632 applyBuilderRules(filter.query.filterBuilder)
633 refresh()
634 }
635 }
636 }
638 private fun applyBuilderRules(builder: FilterBuilder?) {
639 if (builder == null) return
640 for (group in builder.rules) {
641 for (rule in group.rules) {
642 if (rule.field.isNotBlank()) {
643 addFilter(rule.field, rule.operator, rule.value)
644 }
645 }
646 }
647 }
649 fun deleteSavedFilter(filterId: String) {
650 viewModelScope.launch {
651 when (repository.deleteFilter(filterId)) {
652 is ApiResult.Success -> {
653 _state.update {
654 it.copy(
655 savedFilters = it.savedFilters.copy(
656 filters = it.savedFilters.filters.filter { f -> f.filterId != filterId },
657 ),
658 )
659 }
660 }
661 is ApiResult.Error -> { /* Silently ignore — will show on next reload */ }
662 }
663 }
664 }
666 /** Parse a display string like "column = value" back into (field, operator, value) */
667 private fun parseFilterDisplay(display: String): Triple<String, String, String> {
668 for (op in listOf("IS NOT NULL", "IS NULL", "ILIKE", "LIKE", "!=", ">=", "<=", "=", ">", "<")) {
669 val idx = display.indexOf(" $op")
670 if (idx >= 0) {
671 val field = display.substring(0, idx)
672 val value = display.substring(idx + op.length + 1).trim()
673 return Triple(field, op, value)
674 }
675 }
676 return Triple(display, "=", "")
677 }
679 override fun onCleared() {
680 super.onCleared()
681 schemaJob?.cancel()
682 schemaJob = null
683 searchJob?.cancel()
684 searchJob = null
685 stopStreaming()
686 }