git.iko.soy/parseable-android
on 0da331f help sign in
parseable-android/app/src/main/java/com/parseable/android/ui/screens/logviewer/LogViewerViewModel.kt
482 lines 18 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.ApiResult
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 LogViewerState(
53 val streamName: String = "",
54 val logs: List<JsonObject> = emptyList(),
55 val columns: List<String> = emptyList(),
56 val isLoading: Boolean = false,
57 val error: String? = null,
58 val selectedTimeRange: TimeRange = TimeRange.LAST_1H,
59 val filters: FilterState = FilterState(),
60 val streaming: StreamingState = StreamingState(),
61 val currentLimit: Int = 500,
62 val hasMore: Boolean = false,
63 val customStartTime: Long? = null,
64 val customEndTime: Long? = null,
65) {
66 // Convenience accessors for backward compatibility with Screen
67 val searchQuery: String get() = filters.searchQuery
68 val activeFilters: List<String> get() = filters.activeFilters
69 val filterClauses: List<String> get() = filters.filterClauses
70 val customSql: String get() = filters.customSql
71 val isStreaming: Boolean get() = streaming.isStreaming
72 val streamingNewCount: Int get() = streaming.streamingNewCount
73 val streamingError: String? get() = streaming.streamingError
74 val currentIntervalMs: Long get() = streaming.currentIntervalMs
75 val isSearching: Boolean get() = filters.isSearching
78@HiltViewModel
79class LogViewerViewModel @Inject constructor(
80 private val repository: ParseableRepository,
81) : ViewModel() {
83 private val _state = MutableStateFlow(LogViewerState())
84 val state: StateFlow<LogViewerState> = _state.asStateFlow()
86 private val dateFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSSSS'+00:00'")
88 private var streamingJob: Job? = null
89 @Volatile private var streamingGeneration: Int = 0
90 @Volatile private var lastSeenTimestamp: String? = null
91 @Volatile private var consecutiveStreamingErrors: Int = 0
92 private var searchJob: Job? = null
93 private var schemaJob: Job? = null
95 companion object {
96 private const val STREAMING_BASE_INTERVAL_MS = 3000L
97 private const val STREAMING_MAX_INTERVAL_MS = 30000L
98 private const val STREAMING_MAX_LOGS = 1000
99 private const val MAX_LOAD_LIMIT = 5000
100 private const val LOAD_MORE_INCREMENT = 500
101 private const val MAX_STREAMING_ERRORS = 5
102 private val ALLOWED_OPERATORS = setOf("=", "!=", "LIKE", "ILIKE", ">", "<", ">=", "<=", "IS NULL", "IS NOT NULL")
103 }
105 fun initialize(streamName: String) {
106 if (_state.value.streamName == streamName) return
107 stopStreaming()
108 _state.update { it.copy(streamName = streamName) }
109 loadSchema(streamName)
110 refresh()
111 }
113 private fun loadSchema(streamName: String) {
114 schemaJob?.cancel()
115 schemaJob = viewModelScope.launch {
116 when (val result = repository.getStreamSchema(streamName)) {
117 is ApiResult.Success -> {
118 val columns = result.data.fields.map { it.name }
119 _state.update { it.copy(columns = columns) }
120 }
121 is ApiResult.Error -> { /* Schema loading failure is non-fatal */ }
122 }
123 }
124 }
126 fun onTimeRangeChange(range: TimeRange) {
127 stopStreaming()
128 _state.update {
129 it.copy(
130 selectedTimeRange = range,
131 currentLimit = 500,
132 customStartTime = if (range != TimeRange.CUSTOM) null else it.customStartTime,
133 customEndTime = if (range != TimeRange.CUSTOM) null else it.customEndTime,
134 )
135 }
136 if (range != TimeRange.CUSTOM) refresh()
137 }
139 fun setCustomTimeRange(startMillis: Long, endMillis: Long) {
140 stopStreaming()
141 _state.update {
142 it.copy(
143 selectedTimeRange = TimeRange.CUSTOM,
144 customStartTime = startMillis,
145 customEndTime = endMillis,
146 currentLimit = 500,
147 )
148 }
149 refresh()
150 }
152 fun onSearchQueryChange(query: String) {
153 _state.update { it.copy(filters = it.filters.copy(searchQuery = query, isSearching = query.isNotBlank())) }
154 searchJob?.cancel()
155 searchJob = viewModelScope.launch {
156 delay(300)
157 _state.update { it.copy(filters = it.filters.copy(isSearching = false)) }
158 refresh()
159 }
160 }
162 fun addFilter(column: String, operator: String, value: String) {
163 if (operator !in ALLOWED_OPERATORS) return
164 val safeColumn = escapeIdentifier(column)
165 val safeValue = escapeSql(value)
166 val clause = when (operator) {
167 "IS NULL", "IS NOT NULL" -> "\"$safeColumn\" $operator"
168 "LIKE", "ILIKE" -> "\"$safeColumn\" $operator '%$safeValue%'"
169 else -> "\"$safeColumn\" $operator '$safeValue'"
170 }
171 val display = when (operator) {
172 "IS NULL", "IS NOT NULL" -> "$column $operator"
173 else -> "$column $operator $value"
174 }
175 _state.update {
176 it.copy(
177 filters = it.filters.copy(
178 filterClauses = it.filters.filterClauses + clause,
179 activeFilters = it.filters.activeFilters + display,
180 ),
181 )
182 }
183 refresh()
184 }
186 fun removeFilter(display: String) {
187 var removed = false
188 _state.update {
189 val index = it.filters.activeFilters.indexOf(display)
190 if (index >= 0 && index < it.filters.filterClauses.size) {
191 removed = true
192 it.copy(
193 filters = it.filters.copy(
194 activeFilters = it.filters.activeFilters.filterIndexed { i, _ -> i != index },
195 filterClauses = it.filters.filterClauses.filterIndexed { i, _ -> i != index },
196 ),
197 )
198 } else {
199 it
200 }
201 }
202 if (removed) refresh()
203 }
205 fun clearFilters() {
206 _state.update {
207 it.copy(filters = it.filters.copy(activeFilters = emptyList(), filterClauses = emptyList()))
208 }
209 refresh()
210 }
212 fun executeCustomSql(sql: String) {
213 val trimmed = sql.trim()
214 if (!trimmed.uppercase().startsWith("SELECT")) {
215 _state.update { it.copy(error = "Only SELECT queries are allowed") }
216 return
217 }
218 // Enforce a LIMIT to prevent OOM from unbounded queries
219 val safeSql = if (!trimmed.uppercase().contains("LIMIT")) {
220 "$trimmed LIMIT $MAX_LOAD_LIMIT"
221 } else {
222 trimmed
223 }
224 _state.update { it.copy(filters = it.filters.copy(customSql = safeSql), currentLimit = 500) }
225 viewModelScope.launch {
226 _state.update { it.copy(isLoading = true, error = null) }
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 }
246 }
248 private fun buildSearchClause(columns: List<String>, searchQuery: String): String? {
249 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%'"
256 }
257 }
259 fun refresh() {
260 if (_state.value.streamName.isEmpty()) return
262 viewModelScope.launch {
263 _state.update {
264 it.copy(
265 isLoading = true,
266 error = null,
267 streaming = it.streaming.copy(streamingError = null),
268 )
269 }
271 // Snapshot state inside the coroutine so we always see the latest filters/columns
272 val current = _state.value
273 val (startTime, endTime) = getTimeRange()
275 // Build WHERE clause from filters + search
276 val clauses = current.filters.filterClauses.toMutableList()
277 if (current.filters.searchQuery.isNotBlank()) {
278 val searchClause = buildSearchClause(current.columns, current.filters.searchQuery)
279 if (searchClause != null) {
280 clauses.add("($searchClause)")
281 } else {
282 _state.update {
283 it.copy(isLoading = false, error = "Search unavailable: stream schema not loaded")
284 }
285 return@launch
286 }
287 }
289 val filterSql = clauses.joinToString(" AND ")
291 when (val result = repository.queryLogs(
292 stream = current.streamName,
293 startTime = startTime,
294 endTime = endTime,
295 filterSql = filterSql,
296 limit = current.currentLimit,
297 )) {
298 is ApiResult.Success -> {
299 _state.update {
300 it.copy(
301 logs = result.data,
302 isLoading = false,
303 hasMore = result.data.size >= it.currentLimit,
304 )
305 }
306 }
307 is ApiResult.Error -> {
308 _state.update {
309 it.copy(isLoading = false, error = result.userMessage)
310 }
311 }
312 }
313 }
314 }
316 fun loadMore() {
317 val current = _state.value
318 if (current.currentLimit >= MAX_LOAD_LIMIT) {
319 _state.update { it.copy(hasMore = false) }
320 return
321 }
322 val newLimit = (current.currentLimit + LOAD_MORE_INCREMENT).coerceAtMost(MAX_LOAD_LIMIT)
323 _state.update { it.copy(currentLimit = newLimit) }
324 refresh()
325 }
327 private fun getTimeRange(): Pair<String, String> {
328 val current = _state.value
329 if (current.selectedTimeRange == TimeRange.CUSTOM &&
330 current.customStartTime != null && current.customEndTime != null
331 ) {
332 val start = ZonedDateTime.ofInstant(
333 java.time.Instant.ofEpochMilli(current.customStartTime), ZoneOffset.UTC
334 )
335 val end = ZonedDateTime.ofInstant(
336 java.time.Instant.ofEpochMilli(current.customEndTime), ZoneOffset.UTC
337 )
338 return Pair(start.format(dateFormatter), end.format(dateFormatter))
339 }
340 val now = ZonedDateTime.now(ZoneOffset.UTC)
341 val start = now.minusMinutes(current.selectedTimeRange.minutes)
342 return Pair(
343 start.format(dateFormatter),
344 now.format(dateFormatter),
345 )
346 }
348 fun toggleStreaming() {
349 if (_state.value.streaming.isStreaming) {
350 stopStreaming()
351 } else {
352 startStreaming()
353 }
354 }
356 private fun startStreaming() {
357 stopStreaming()
358 consecutiveStreamingErrors = 0
359 val generation = ++streamingGeneration
360 _state.update {
361 it.copy(streaming = StreamingState(isStreaming = true, streamingNewCount = 0))
362 }
364 // Set the baseline timestamp to "now" so we only poll for new logs
365 lastSeenTimestamp = ZonedDateTime.now(ZoneOffset.UTC).format(dateFormatter)
367 streamingJob = viewModelScope.launch {
368 while (isActive && streamingGeneration == generation) {
369 pollNewLogs()
370 // Exponential backoff on errors, normal interval on success
371 val intervalMs = if (consecutiveStreamingErrors > 0) {
372 (STREAMING_BASE_INTERVAL_MS * (1L shl consecutiveStreamingErrors.coerceAtMost(4)))
373 .coerceAtMost(STREAMING_MAX_INTERVAL_MS)
374 } else {
375 STREAMING_BASE_INTERVAL_MS
376 }
377 _state.update {
378 it.copy(streaming = it.streaming.copy(currentIntervalMs = intervalMs))
379 }
380 delay(intervalMs)
381 }
382 }
383 }
385 fun stopStreaming() {
386 streamingGeneration++
387 streamingJob?.cancel()
388 streamingJob = null
389 _state.update {
390 it.copy(streaming = it.streaming.copy(isStreaming = false))
391 }
392 }
394 fun dismissStreamingError() {
395 _state.update {
396 it.copy(streaming = it.streaming.copy(streamingError = null))
397 }
398 }
400 private suspend fun pollNewLogs() {
401 val current = _state.value
402 if (current.streamName.isEmpty()) return
404 val startTime = lastSeenTimestamp ?: return
405 val endTime = ZonedDateTime.now(ZoneOffset.UTC).format(dateFormatter)
407 // Build WHERE clause from active filters + search
408 val clauses = current.filters.filterClauses.toMutableList()
409 if (current.filters.searchQuery.isNotBlank()) {
410 val searchClause = buildSearchClause(current.columns, current.filters.searchQuery)
411 if (searchClause != null) {
412 clauses.add(searchClause)
413 }
414 }
416 val whereClause = if (clauses.isNotEmpty()) " WHERE ${clauses.joinToString(" AND ")}" else ""
417 val safeName = escapeIdentifier(current.streamName)
418 val sql = "SELECT * FROM \"$safeName\"$whereClause ORDER BY p_timestamp DESC LIMIT 200"
420 when (val result = repository.queryLogsRaw(sql, startTime, endTime)) {
421 is ApiResult.Success -> {
422 val newLogs = result.data
423 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
428 } catch (_: Exception) {
429 null
430 }
431 if (newestTimestamp != null) {
432 lastSeenTimestamp = newestTimestamp
433 }
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
442 }
443 state.copy(
444 logs = capped,
445 streaming = state.streaming.copy(
446 streamingNewCount = state.streaming.streamingNewCount + newLogs.size,
447 streamingError = null,
448 ),
449 )
450 }
451 }
452 consecutiveStreamingErrors = 0
453 }
454 is ApiResult.Error -> {
455 consecutiveStreamingErrors++
456 if (consecutiveStreamingErrors >= MAX_STREAMING_ERRORS) {
457 stopStreaming()
458 _state.update {
459 it.copy(error = "Streaming stopped after repeated errors: ${result.userMessage}")
460 }
461 } else if (consecutiveStreamingErrors >= 3) {
462 _state.update {
463 it.copy(
464 streaming = it.streaming.copy(
465 streamingError = "Streaming interrupted: ${result.userMessage}. Retrying..."
466 ),
467 )
468 }
469 }
470 }
471 }
472 }
474 override fun onCleared() {
475 super.onCleared()
476 schemaJob?.cancel()
477 schemaJob = null
478 searchJob?.cancel()
479 searchJob = null
480 stopStreaming()
481 }