Flow on the phone: a repository's runs, a run's jobs, live logs, rerun and cancel (refs #5) 1c42a70fcba3

package io.oxygenhub.android.ui.flow

import androidx.compose.foundation.clickable

import androidx.compose.foundation.horizontalScroll

import androidx.compose.foundation.layout.Arrangement

import androidx.compose.foundation.layout.Column

import androidx.compose.foundation.layout.Row

import androidx.compose.foundation.layout.fillMaxSize

import androidx.compose.foundation.layout.fillMaxWidth

import androidx.compose.foundation.layout.padding

import androidx.compose.foundation.layout.width

import androidx.compose.foundation.lazy.LazyColumn

import androidx.compose.foundation.lazy.items

import androidx.compose.foundation.lazy.rememberLazyListState

import androidx.compose.foundation.rememberScrollState

import androidx.compose.foundation.text.selection.SelectionContainer

import androidx.compose.material.icons.Icons

import androidx.compose.material.icons.outlined.Block

import androidx.compose.material.icons.outlined.CheckCircle

import androidx.compose.material.icons.outlined.Error

import androidx.compose.material.icons.outlined.HourglassEmpty

import androidx.compose.material.icons.outlined.PlayCircle

import androidx.compose.material.icons.outlined.RemoveCircleOutline

import androidx.compose.material.icons.outlined.Replay

import androidx.compose.material.icons.outlined.RocketLaunch

import androidx.compose.material3.HorizontalDivider

import androidx.compose.material3.Icon

import androidx.compose.material3.IconButton

import androidx.compose.material3.ListItem

import androidx.compose.material3.MaterialTheme

import androidx.compose.material3.Text

import androidx.compose.runtime.Composable

import androidx.compose.runtime.LaunchedEffect

import androidx.compose.runtime.getValue

import androidx.compose.ui.Modifier

import androidx.compose.ui.graphics.Color

import androidx.compose.ui.graphics.vector.ImageVector

import androidx.compose.ui.text.font.FontFamily

import androidx.compose.ui.unit.dp

import androidx.lifecycle.compose.collectAsStateWithLifecycle

import io.oxygenhub.android.api.FlowJob

import io.oxygenhub.android.api.FlowRun

import io.oxygenhub.android.ui.common.AppScreen

import io.oxygenhub.android.ui.common.EmptyState

import io.oxygenhub.android.ui.common.ErrorState

import io.oxygenhub.android.ui.common.Load

import io.oxygenhub.android.ui.common.LoadContent

import io.oxygenhub.android.ui.common.LoadingState

import io.oxygenhub.android.ui.common.relativeTime

import io.oxygenhub.android.ui.theme.LocalStatusColors

/** A status's icon and colour, as the hub's run pages show them. */

@Composable

fun statusLook(status: String): Pair<ImageVector, Color> {

    val c = LocalStatusColors.current

    return when (status) {

        "passed" -> Icons.Outlined.CheckCircle to c.ok

        "failed", "error" -> Icons.Outlined.Error to c.bad

        "running" -> Icons.Outlined.PlayCircle to c.warn

        "cancelled" -> Icons.Outlined.Block to MaterialTheme.colorScheme.onSurfaceVariant

        "skipped" -> Icons.Outlined.RemoveCircleOutline to MaterialTheme.colorScheme.onSurfaceVariant

        else -> Icons.Outlined.HourglassEmpty to MaterialTheme.colorScheme.onSurfaceVariant

    }

}

/** A repository's runs, for its tab. */

@Composable

fun RunList(vm: RunsViewModel, onOpen: (FlowRun) -> Unit) {

    val state by vm.state.collectAsStateWithLifecycle()

    LoadContent(state, onRefresh = vm::refresh) { runs ->

        if (runs.isEmpty()) {

            EmptyState(Icons.Outlined.RocketLaunch, "No runs yet", "A push to ${vm.owner}/${vm.name} with a .flow/pipeline.kdl starts one.")

        } else {

            LazyColumn(Modifier.fillMaxSize()) {

                items(runs, key = { it.number }) { r ->

                    val (icon, colour) = statusLook(r.status)

                    ListItem(

                        modifier = Modifier.clickable { onOpen(r) },

                        leadingContent = { Icon(icon, contentDescription = r.status, tint = colour) },

                        headlineContent = { Text("#${r.number} ${r.bookmark}") },

                        supportingContent = {

                            Text(

                                listOfNotNull(

                                    r.status,

                                    "${r.trigger} by ${r.actor}",

                                    relativeTime(r.createdMs).takeIf { it.isNotEmpty() },

                                    r.queuePosition?.let { "#$it in the queue" },

                                ).joinToString(" · "),

                            )

                        },

                    )

                    HorizontalDivider()

                }

            }

        }

    }

}

/** A run: its jobs, refreshed while it is live; rerun or cancel it. */

@Composable

fun RunScreen(vm: RunViewModel, canWrite: Boolean, onBack: () -> Unit, onOpenJob: (FlowJob) -> Unit) {

    val state by vm.state.collectAsStateWithLifecycle()

    val error by vm.error.collectAsStateWithLifecycle()

    val run = (state as? Load.Ready)?.value

    AppScreen(

        title = "Run #${vm.number}",

        subtitle = listOfNotNull("${vm.owner}/${vm.name}", run?.bookmark, run?.status).joinToString(" · "),

        onBack = onBack,

        actions = {

            if (canWrite && run != null) {

                if (run.live) {

                    IconButton(onClick = vm::cancel) { Icon(Icons.Outlined.Block, contentDescription = "Cancel the run") }

                } else {

                    IconButton(onClick = vm::rerun) { Icon(Icons.Outlined.Replay, contentDescription = "Run it again") }

                }

            }

        },

    ) {

        LoadContent(state, onRefresh = vm::refresh) { r ->

            LazyColumn(Modifier.fillMaxSize()) {

                if (r.error.isNotEmpty()) item { Text(r.error, color = MaterialTheme.colorScheme.error, modifier = Modifier.padding(16.dp)) }

                error?.let { item { Text(it, color = MaterialTheme.colorScheme.error, modifier = Modifier.padding(16.dp)) } }

                items(r.jobs, key = { it.name }) { j ->

                    val (icon, colour) = statusLook(j.status)

                    ListItem(

                        modifier = Modifier.clickable { onOpenJob(j) },

                        leadingContent = { Icon(icon, contentDescription = j.status, tint = colour) },

                        headlineContent = { Text(j.name) },

                        supportingContent = {

                            Column {

                                Text(

                                    listOfNotNull(

                                        j.status,

                                        j.seconds()?.let { "$it s" },

                                        j.runner.takeIf { it.isNotEmpty() }?.let { "on $it" },

                                        j.exitCode?.let { "exit $it" },

                                    ).joinToString(" · "),

                                )

                                if (j.waiting.isNotEmpty()) Text(j.waiting, style = MaterialTheme.typography.bodySmall)

                                if (j.error.isNotEmpty()) Text(j.error, style = MaterialTheme.typography.bodySmall, color = MaterialTheme.colorScheme.error)

                            }

                        },

                        trailingContent = if (canWrite && !j.live && !r.live) {

                            { IconButton(onClick = { vm.rerunJob(j.name) }) { Icon(Icons.Outlined.Replay, contentDescription = "Run ${j.name} again") } }

                        } else {

                            null

                        },

                    )

                    HorizontalDivider()

                }

            }

        }

    }

}

/** A job's log, following new lines while it runs. */

@Composable

fun JobLogScreen(vm: JobLogViewModel, onBack: () -> Unit) {

    val state by vm.state.collectAsStateWithLifecycle()

    val list = rememberLazyListState()

    val status = LocalStatusColors.current

    val log = (state as? Load.Ready)?.value

    // Follow the end while the job runs.

    LaunchedEffect(log?.lines?.size) {

        if (log != null && log.live && log.lines.isNotEmpty()) list.scrollToItem(log.lines.size - 1)

    }

    AppScreen(title = vm.job, subtitle = listOfNotNull("${vm.owner}/${vm.name} #${vm.number}", log?.status).joinToString(" · "), onBack = onBack) {

        when (val s = state) {

            Load.Loading -> LoadingState()

            is Load.Failed -> ErrorState(s.message, onRetry = vm::retry)

            is Load.Ready -> SelectionContainer {

                LazyColumn(Modifier.fillMaxSize().horizontalScroll(rememberScrollState()), state = list) {

                    items(s.value.lines, key = { it.seq }) { l ->

                        Row(Modifier.padding(horizontal = 8.dp), horizontalArrangement = Arrangement.spacedBy(8.dp)) {

                            Text(

                                "%.1fs".format(l.t / 1000.0),

                                modifier = Modifier.width(56.dp),

                                fontFamily = FontFamily.Monospace,

                                style = MaterialTheme.typography.bodySmall,

                                color = MaterialTheme.colorScheme.outline,

                            )

                            Text(

                                l.text,

                                fontFamily = FontFamily.Monospace,

                                style = MaterialTheme.typography.bodySmall,

                                softWrap = false,

                                color = when (l.stream) {

                                    "stderr" -> status.bad

                                    "flow" -> MaterialTheme.colorScheme.primary

                                    else -> MaterialTheme.colorScheme.onSurface

                                },

                            )

                        }

                    }

                    if (s.value.live) item { Text("…", modifier = Modifier.fillMaxWidth().padding(8.dp)) }

                }

            }

        }

    }

}

package io.oxygenhub.android.ui.flow

import androidx.lifecycle.ViewModel

import androidx.lifecycle.viewModelScope

import io.oxygenhub.android.api.FlowRun

import io.oxygenhub.android.api.HubApi

import io.oxygenhub.android.api.LogLine

import io.oxygenhub.android.api.userMessage

import io.oxygenhub.android.ui.common.Load

import io.oxygenhub.android.ui.common.load

import kotlinx.coroutines.CancellationException

import kotlinx.coroutines.Job

import kotlinx.coroutines.delay

import kotlinx.coroutines.flow.MutableStateFlow

import kotlinx.coroutines.flow.StateFlow

import kotlinx.coroutines.flow.asStateFlow

import kotlinx.coroutines.isActive

import kotlinx.coroutines.launch

/** A repository's runs, newest first (#5). */

class RunsViewModel(private val api: HubApi, val owner: String, val name: String) : ViewModel() {

    private val _state = MutableStateFlow<Load<List<FlowRun>>>(Load.Loading)

    val state: StateFlow<Load<List<FlowRun>>> = _state.asStateFlow()

    private var job: Job? = null

    init {

        refresh()

    }

    fun refresh() {

        job = load(_state, job) { api.runs(owner, name) }

    }

}

/**

 * One run and its jobs, fetched again every few seconds while it is queued

 * or running, so its jobs' states follow; rerun and cancel it.

 */

class RunViewModel(

    private val api: HubApi,

    val owner: String,

    val name: String,

    val number: Long,

    /** How often a live run is fetched again; 0 not at all (tests). */

    private val pollMs: Long = 3000,

) : ViewModel() {

    private val _state = MutableStateFlow<Load<FlowRun>>(Load.Loading)

    val state: StateFlow<Load<FlowRun>> = _state.asStateFlow()

    private val _error = MutableStateFlow<String?>(null)

    val error: StateFlow<String?> = _error.asStateFlow()

    private var job: Job? = null

    init {

        refresh()

        if (pollMs > 0) {

            viewModelScope.launch {

                while (isActive) {

                    delay(pollMs)

                    if ((_state.value as? Load.Ready)?.value?.live == true) poll()

                }

            }

        }

    }

    fun refresh() {

        job = load(_state, job) { api.run(owner, name, number) }

    }

    /** Fetch it again without a spinner. */

    suspend fun poll() {

        try {

            _state.value = Load.Ready(api.run(owner, name, number))

        } catch (e: CancellationException) {

            throw e

        } catch (_: Exception) {

            // The next tick tries again; the last state stays shown.

        }

    }

    fun rerun() = act { api.rerunRun(owner, name, number) }

    fun rerunJob(job: String) = act { api.rerunJob(owner, name, number, job) }

    fun cancel() = act { api.cancelRun(owner, name, number) }

    private fun act(step: suspend () -> Unit) {

        _error.value = null

        viewModelScope.launch {

            try {

                step()

                poll()

            } catch (e: CancellationException) {

                throw e

            } catch (e: Exception) {

                _error.value = e.userMessage()

            }

        }

    }

}

/**

 * A job's log, followed while the job runs: new lines are asked for after

 * the last one held, every couple of seconds, until the job has finished.

 */

class JobLogViewModel(

    private val api: HubApi,

    val owner: String,

    val name: String,

    val number: Long,

    val job: String,

    private val pollMs: Long = 2000,

) : ViewModel() {

    data class Log(val lines: List<LogLine>, val status: String, val live: Boolean)

    private val _state = MutableStateFlow<Load<Log>>(Load.Loading)

    val state: StateFlow<Load<Log>> = _state.asStateFlow()

    init {

        viewModelScope.launch {

            next()

            while (pollMs > 0 && isActive && (_state.value as? Load.Ready)?.value?.live != false) {

                delay(pollMs)

                next()

            }

        }

    }

    /** The lines after the last one held, and whether the job still runs. */

    suspend fun next() {

        val held = (_state.value as? Load.Ready)?.value

        try {

            val after = held?.lines?.lastOrNull()?.seq ?: 0

            val more = api.jobLog(owner, name, number, job, after)

            val j = api.run(owner, name, number).jobs.firstOrNull { it.name == job }

            _state.value = Load.Ready(Log((held?.lines.orEmpty()) + more, j?.status ?: held?.status.orEmpty(), j?.live ?: false))

        } catch (e: CancellationException) {

            throw e

        } catch (e: Exception) {

            if (held == null) _state.value = Load.failed(e)

        }

    }

    fun retry() {

        _state.value = Load.Loading

        viewModelScope.launch { next() }

    }

}

package io.oxygenhub.android.ui

import io.oxygenhub.android.FakeHubApi

import io.oxygenhub.android.MainDispatcherRule

import io.oxygenhub.android.api.FlowJob

import io.oxygenhub.android.api.FlowRun

import io.oxygenhub.android.api.LogLine

import io.oxygenhub.android.ui.common.Load

import io.oxygenhub.android.ui.flow.JobLogViewModel

import io.oxygenhub.android.ui.flow.RunViewModel

import io.oxygenhub.android.ui.flow.RunsViewModel

import kotlinx.coroutines.test.runTest

import org.junit.Assert.assertEquals

import org.junit.Assert.assertFalse

import org.junit.Assert.assertTrue

import org.junit.Rule

import org.junit.Test

class FlowViewModelsTest {

    @get:Rule val main = MainDispatcherRule()

    private val api = FakeHubApi().apply {

        runList = listOf(FlowRun(56, "main", "passed"), FlowRun(55, "main", "failed"))

        runStore[56] = FlowRun(56, "main", "running", jobs = listOf(FlowJob("count", "running", runner = "android-pixel-10-pro-fold")))

        logs["count"] = mutableListOf(LogLine(1, 0, "flow", "fetching"), LogLine(2, 1800, "stdout", "step 1 of 120"))

    }

    @Test

    fun listsTheRepositorysRuns() {

        val runs = (RunsViewModel(api, "andymccall", "phone-runner-demo").state.value as Load.Ready).value

        assertEquals(listOf(56L, 55L), runs.map { it.number })

    }

    @Test

    fun aRunFollowsItsJobsAndCanBeCancelled() = runTest {

        val vm = RunViewModel(api, "andymccall", "phone-runner-demo", 56, pollMs = 0)

        assertTrue((vm.state.value as Load.Ready).value.live)

        api.runStore[56] = api.runStore.getValue(56).copy(status = "passed", jobs = listOf(FlowJob("count", "passed", runner = "plex-1")))

        vm.poll()

        assertEquals("plex-1", (vm.state.value as Load.Ready).value.jobs[0].runner)

        vm.cancel()

        assertTrue("cancel 56" in api.calls)

        vm.rerunJob("count")

        assertTrue("rerun 56 count" in api.calls)

    }

    @Test

    fun aLogGrowsFromTheLastLineHeldUntilTheJobEnds() = runTest {

        val vm = JobLogViewModel(api, "andymccall", "phone-runner-demo", 56, "count", pollMs = 0)

        var log = (vm.state.value as Load.Ready).value

        assertEquals(listOf("fetching", "step 1 of 120"), log.lines.map { it.text })

        assertTrue(log.live)

        api.logs.getValue("count") += LogLine(3, 2500, "flow", "── handed back by android-pixel-10-pro-fold: off Wi-Fi")

        api.runStore[56] = api.runStore.getValue(56).copy(status = "running", jobs = listOf(FlowJob("count", "queued")))

        vm.next()

        log = (vm.state.value as Load.Ready).value

        assertEquals(3, log.lines.size)

        assertTrue("log count after 2" in api.calls)

        api.runStore[56] = api.runStore.getValue(56).copy(status = "passed", jobs = listOf(FlowJob("count", "passed")))

        vm.next()

        assertFalse((vm.state.value as Load.Ready).value.live)

        assertEquals("passed", (vm.state.value as Load.Ready).value.status)

    }

}

    /** A repository's runs, newest first. */

    suspend fun runs(owner: String, name: String): List<FlowRun>

    /** One run, with its jobs. */

    suspend fun run(owner: String, name: String, number: Long): FlowRun

    /** A job's log lines after [after] (0: from the start). */

    suspend fun jobLog(owner: String, name: String, number: Long, job: String, after: Long): List<LogLine>

    suspend fun rerunRun(owner: String, name: String, number: Long)

    /** Run [job] again in the same run, with the jobs that depend on it. */

    suspend fun rerunJob(owner: String, name: String, number: Long, job: String)

    suspend fun cancelRun(owner: String, name: String, number: Long)

    override suspend fun runs(owner: String, name: String): List<FlowRun> =

        get(url("repos", owner, name, "flow", "runs"), ListSerializer(FlowRun.serializer()))

    override suspend fun run(owner: String, name: String, number: Long): FlowRun =

        get(url("repos", owner, name, "flow", "runs", number.toString()), FlowRun.serializer())

    override suspend fun jobLog(owner: String, name: String, number: Long, job: String, after: Long): List<LogLine> =

        get(

            url("repos", owner, name, "flow", "runs", number.toString(), "jobs", job, "log") { addQueryParameter("after", after.toString()) },

            ListSerializer(LogLine.serializer()),

        )

    override suspend fun rerunRun(owner: String, name: String, number: Long) {

        send("POST", url("repos", owner, name, "flow", "runs", number.toString(), "rerun"))

    }

    override suspend fun rerunJob(owner: String, name: String, number: Long, job: String) {

        send("POST", url("repos", owner, name, "flow", "runs", number.toString(), "jobs", job, "rerun"))

    }

    override suspend fun cancelRun(owner: String, name: String, number: Long) {

        send("POST", url("repos", owner, name, "flow", "runs", number.toString(), "cancel"))

    }

/** A Flow run of a repository's pipeline (#5). */

@Serializable

data class FlowRun(

    val number: Long,

    val bookmark: String = "",

    /** `queued`, `running`, `passed`, `failed`, `error` or `cancelled`. */

    val status: String = "queued",

    /** `push`, `manual`, `schedule`, `proposal` or `rerun:N`. */

    val trigger: String = "",

    val actor: String = "",

    val error: String = "",

    @SerialName("created_ms") val createdMs: Long = 0,

    @SerialName("finished_ms") val finishedMs: Long? = null,

    val url: String = "",

    @SerialName("queue_position") val queuePosition: Long? = null,

    val jobs: List<FlowJob> = emptyList(),

) {

    val live: Boolean get() = status == "queued" || status == "running"

}

/** A job of a run. */

@Serializable

data class FlowJob(

    val name: String,

    val status: String = "queued",

    @SerialName("depends_on") val dependsOn: List<String> = emptyList(),

    val runner: String = "",

    @SerialName("exit_code") val exitCode: Long? = null,

    val error: String = "",

    @SerialName("started_ms") val startedMs: Long? = null,

    @SerialName("finished_ms") val finishedMs: Long? = null,

    @SerialName("log_lines") val logLines: Long = 0,

    val waiting: String = "",

) {

    val live: Boolean get() = status == "queued" || status == "running"

    /** How long it ran, or has been running, in seconds. */

    fun seconds(now: Long = System.currentTimeMillis()): Long? = startedMs?.let { ((finishedMs ?: now) - it) / 1000 }

}

/** A line of a job's log: `stdout`, `stderr` or `flow` (the runner's own). */

@Serializable

data class LogLine(val seq: Long, val t: Long = 0, val stream: String = "stdout", val text: String = "")

import io.oxygenhub.android.ui.flow.JobLogScreen

import io.oxygenhub.android.ui.flow.JobLogViewModel

import io.oxygenhub.android.ui.flow.RunScreen

import io.oxygenhub.android.ui.flow.RunViewModel

import io.oxygenhub.android.ui.flow.RunsViewModel

/** A Flow run, and a job's log (#5). */

@Serializable data class RunRoute(val owner: String, val name: String, val number: Long)

@Serializable data class JobLogRoute(val owner: String, val name: String, val number: Long, val job: String)

                        rememberViewModel(key = "runs:$slug") { RunsViewModel(api, owner, name) },

                        onOpenRun = { nav.navigate(RunRoute(owner, name, it)) },

                    rememberViewModel { RunsViewModel(api, r.owner, r.name) },

                    onOpenRun = { nav.navigate(RunRoute(r.owner, r.name, it)) },

            }

            composable<RunRoute> { e ->

                val r = e.toRoute<RunRoute>()

                RunScreen(

                    rememberViewModel { RunViewModel(api, r.owner, r.name, r.number) },

                    canWrite = session.canWrite,

                    onBack = { nav.popBackStack() },

                    onOpenJob = { j -> nav.navigate(JobLogRoute(r.owner, r.name, r.number, j.name)) },

                )

            }

            composable<JobLogRoute> { e ->

                val r = e.toRoute<JobLogRoute>()

                JobLogScreen(rememberViewModel { JobLogViewModel(api, r.owner, r.name, r.number, r.job) }, onBack = { nav.popBackStack() })

import androidx.compose.material3.PrimaryScrollableTabRow

import androidx.compose.ui.unit.dp

import io.oxygenhub.android.ui.flow.RunList

import io.oxygenhub.android.ui.flow.RunsViewModel

/** A repository's tabs: its code, issues, proposals and Flow runs (#3, #5). */

enum class RepositoryTab(val label: String) { Code("Code"), Issues("Issues"), Proposals("Proposals"), Runs("Runs") }

    runs: RunsViewModel,

    onOpenRun: (Long) -> Unit,

            PrimaryScrollableTabRow(selectedTabIndex = tab, edgePadding = 0.dp) {

                RepositoryTab.Runs -> RunList(runs) { onOpenRun(it.number) }

    var runList: List<io.oxygenhub.android.api.FlowRun> = emptyList()

    override suspend fun runs(owner: String, name: String) = runList.also { calls += "runs" }

    val runStore = mutableMapOf<Long, io.oxygenhub.android.api.FlowRun>()

    override suspend fun run(owner: String, name: String, number: Long) = runStore.getValue(number).also { calls += "run $number" }

    val logs = mutableMapOf<String, MutableList<io.oxygenhub.android.api.LogLine>>()

    override suspend fun jobLog(owner: String, name: String, number: Long, job: String, after: Long) =

        logs[job].orEmpty().filter { it.seq > after }.also { calls += "log $job after $after" }

    override suspend fun rerunRun(owner: String, name: String, number: Long) {

        calls += "rerun $number"

    }

    override suspend fun rerunJob(owner: String, name: String, number: Long, job: String) {

        calls += "rerun $number $job"

    }

    override suspend fun cancelRun(owner: String, name: String, number: Long) {

        calls += "cancel $number"

        runStore[number]?.let { runStore[number] = it.copy(status = "cancelled") }

    }