feat(android): обновлять список альбомов в реальном времени

Приложение подписывается на персональный поток GET /api/sse/me и применяет
album.created/updated/deleted к кэшу Room, поэтому список и открытый экран
следуют изменениям с другого устройства или из веба без ручного обновления.

- PersonalSseClient и парсер событий жизненного цикла; подписка живёт вместе с
  сессией (AlbumsViewModel), закрывается на выходе
- AlbumRepository.apply: upsert альбома или удаление его вместе с фото;
  при переподключении список догоняется серверным снимком
- потеря доступа (удаление альбома или исключение участника) убирает альбом из
  списка, а открытый экран возвращается к списку
- README: раздел Live updates и ограничение с истечением токена
This commit is contained in:
iu
2026-10-16 12:47:53 +03:00
parent 44cb7c7d68
commit 9061f7fc44
23 changed files with 906 additions and 13 deletions
@@ -300,6 +300,17 @@ class GaloApi(
.header("Accept", "text/event-stream")
.build()
/**
* Raw authenticated request for the personal stream (album lifecycle events
* for the signed-in user). Authorization travels in the header, so unlike the
* web client no `token` query parameter is needed.
*/
fun sseMeRequest(): Request =
Request.Builder()
.url(api.endpoint("api/sse/me"))
.header("Accept", "text/event-stream")
.build()
private suspend fun get(path: String): HttpResult =
execute(Request.Builder().url(api.endpoint(path)).get().build())
@@ -0,0 +1,42 @@
package com.galo.android.data.network
import com.galo.android.data.network.dto.AlbumDto
import kotlinx.serialization.json.jsonObject
import kotlinx.serialization.json.jsonPrimitive
/** Album lifecycle events delivered on the personal stream (`GET /api/sse/me`). */
sealed interface PersonalEvent {
data class AlbumUpserted(val album: AlbumDto) : PersonalEvent
data class AlbumDeleted(val albumId: String) : PersonalEvent
/** Emitted after a dropped connection so the caller can catch up over REST. */
data object Reconnected : PersonalEvent
}
/**
* Maps the personal stream's `event:`/`data:` pairs to typed [PersonalEvent]s.
*
* Only album lifecycle events affect the local album list. Notifications carry an
* unrelated `type` inside their payload and are ignored, as are unknown event
* types. An event whose payload yields no album id is treated as malformed.
*/
object PersonalEventParser {
fun parse(type: String, data: String): PersonalEvent? = when (type) {
"album.created", "album.updated" ->
runCatching { GaloJson.decodeFromString<AlbumDto>(data) }
.getOrNull()
?.takeIf { it.id.isNotBlank() }
?.let { PersonalEvent.AlbumUpserted(it) }
"album.deleted" -> stringField(data, "id")
?.takeIf { it.isNotBlank() }
?.let { PersonalEvent.AlbumDeleted(it) }
else -> null
}
private fun stringField(data: String, field: String): String? = runCatching {
GaloJson.parseToJsonElement(data).jsonObject[field]?.jsonPrimitive?.content
}.getOrNull()
}
@@ -0,0 +1,76 @@
package com.galo.android.data.network
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.channelFlow
import okhttp3.OkHttpClient
import okhttp3.Response
import okhttp3.sse.EventSource
import okhttp3.sse.EventSourceListener
import okhttp3.sse.EventSources
/**
* Album lifecycle events over the personal SSE stream (`GET /api/sse/me`).
* Reconnects with exponential backoff and emits [PersonalEvent.Reconnected] after
* each drop so the consumer can catch up over REST.
*/
class PersonalSseClient(
private val api: GaloApi,
private val client: OkHttpClient,
private val factory: EventSource.Factory = EventSources.createFactory(client),
) {
fun events(): Flow<PersonalEvent> = channelFlow {
var backoff = INITIAL_BACKOFF_MS
while (true) {
val request = runCatching { api.sseMeRequest() }.getOrNull() ?: break
val closed = CompletableDeferred<Unit>()
val listener = object : EventSourceListener() {
override fun onOpen(eventSource: EventSource, response: Response) {
backoff = INITIAL_BACKOFF_MS
}
override fun onEvent(
eventSource: EventSource,
id: String?,
type: String?,
data: String,
) {
if (type.isNullOrBlank()) return
PersonalEventParser.parse(type, data)?.let { trySend(it) }
}
override fun onClosed(eventSource: EventSource) {
closed.complete(Unit)
}
override fun onFailure(
eventSource: EventSource,
t: Throwable?,
response: Response?,
) {
closed.complete(Unit)
}
}
val source = factory.newEventSource(request, listener)
try {
closed.await()
} finally {
// Runs on a drop and, just as importantly, when the collector is
// cancelled: the connection must not outlive its subscription.
source.cancel()
}
if (isClosedForSend) break
trySend(PersonalEvent.Reconnected)
delay(backoff)
backoff = (backoff * 2).coerceAtMost(MAX_BACKOFF_MS)
}
}
private companion object {
const val INITIAL_BACKOFF_MS = 1_000L
const val MAX_BACKOFF_MS = 30_000L
}
}
@@ -0,0 +1,28 @@
package com.galo.android.data.repo
import com.galo.android.data.network.PersonalEvent
import kotlinx.coroutines.flow.Flow
/**
* Applies the personal album stream to the cache. Started once per session and
* cancelled with the scope that runs it, so its lifetime follows the session.
*/
interface LiveAlbumSource {
suspend fun run()
}
/**
* Bridges the personal SSE stream to the album cache: every event is applied to
* [AlbumRepository]. The container builds one of these, and the UI only starts
* and stops collecting it.
*/
class AlbumLiveSync(
private val albums: AlbumRepository,
private val events: Flow<PersonalEvent>,
) : LiveAlbumSource {
/** Collects the stream until cancelled, applying every event to the cache. */
override suspend fun run() {
events.collect { albums.apply(it) }
}
}
@@ -3,6 +3,7 @@ package com.galo.android.data.repo
import com.galo.android.data.mapper.toDomain
import com.galo.android.data.mapper.toEntity
import com.galo.android.data.network.ObjectUrlResolver
import com.galo.android.data.network.PersonalEvent
import com.galo.android.domain.model.Album
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.map
@@ -15,6 +16,7 @@ import kotlinx.coroutines.flow.map
class AlbumRepository(
private val remote: AlbumRemoteSource,
private val local: AlbumLocalSource,
private val photos: PhotoLocalSource,
private val resolver: ObjectUrlResolver,
) {
@@ -40,4 +42,24 @@ class AlbumRepository(
} catch (_: Exception) {
false
}
/**
* Applies an album lifecycle event from the personal stream to the cache.
* A created or updated album replaces the cached one from the payload without
* a list request; a deleted album is dropped together with its cached photos
* (there is no foreign key to cascade). After a reconnect the list is re-synced
* so changes missed while the stream was down are reflected.
*/
suspend fun apply(event: PersonalEvent) {
when (event) {
is PersonalEvent.AlbumUpserted -> local.upsert(listOf(event.album.toEntity()))
is PersonalEvent.AlbumDeleted -> {
local.deleteByIds(listOf(event.albumId))
photos.clearAlbum(event.albumId)
}
PersonalEvent.Reconnected -> refresh()
}
}
}
@@ -19,6 +19,7 @@ interface AlbumLocalSource {
fun observeEntities(): Flow<List<AlbumEntity>>
suspend fun replaceAll(albums: List<AlbumEntity>)
suspend fun upsert(albums: List<AlbumEntity>)
suspend fun deleteByIds(ids: List<String>)
suspend fun count(): Int
}
@@ -32,6 +33,7 @@ interface PhotoLocalSource {
fun observeForAlbum(albumId: String): Flow<List<PhotoEntity>>
suspend fun upsertAll(photos: List<PhotoEntity>)
suspend fun deleteById(id: String)
suspend fun clearAlbum(albumId: String)
suspend fun countForAlbum(albumId: String): Int
}
@@ -44,6 +46,7 @@ class RoomAlbumLocalSource(private val dao: AlbumDao) : AlbumLocalSource {
override fun observeEntities(): Flow<List<AlbumEntity>> = dao.observeAll()
override suspend fun replaceAll(albums: List<AlbumEntity>) = dao.replaceAll(albums)
override suspend fun upsert(albums: List<AlbumEntity>) = dao.upsertAll(albums)
override suspend fun deleteByIds(ids: List<String>) = dao.deleteByIds(ids)
override suspend fun count(): Int = dao.count()
}
@@ -65,5 +68,7 @@ class RoomPhotoLocalSource(private val dao: PhotoDao) : PhotoLocalSource {
override suspend fun deleteById(id: String) = dao.deleteById(id)
override suspend fun clearAlbum(albumId: String) = dao.clearAlbum(albumId)
override suspend fun countForAlbum(albumId: String): Int = dao.countForAlbum(albumId)
}
@@ -11,8 +11,10 @@ import com.galo.android.data.network.AlbumSseClient
import com.galo.android.data.network.ApiClient
import com.galo.android.data.network.GaloApi
import com.galo.android.data.network.HealthChecker
import com.galo.android.data.network.PersonalSseClient
import com.galo.android.data.prefs.SettingsRepository
import com.galo.android.data.prefs.SettingsState
import com.galo.android.data.repo.AlbumLiveSync
import com.galo.android.data.repo.AlbumRepository
import com.galo.android.data.repo.ApiAlbumRemoteSource
import com.galo.android.data.repo.ApiPhotoRemoteSource
@@ -62,21 +64,28 @@ class AppContainer(context: Context) {
val api = GaloApi(apiClient)
val healthChecker = HealthChecker(apiClient.client)
val albumSseClient = AlbumSseClient(api, apiClient.sseClient)
val personalSseClient = PersonalSseClient(api, apiClient.sseClient)
private val database: GaloDatabase = GaloDatabase.create(appContext)
private val photoLocal = RoomPhotoLocalSource(database.photoDao())
val albumRepository = AlbumRepository(
remote = ApiAlbumRemoteSource(api),
local = RoomAlbumLocalSource(database.albumDao()),
photos = photoLocal,
resolver = apiClient.objectUrls,
)
val photoRepository = PhotoRepository(
remote = ApiPhotoRemoteSource(api),
local = RoomPhotoLocalSource(database.photoDao()),
local = photoLocal,
resolver = apiClient.objectUrls,
)
/** Personal stream applied to the album cache; collected while the session lives. */
val albumLiveSync = AlbumLiveSync(albumRepository, personalSseClient.events())
val uploadRepository = UploadRepository(appContext, database.uploadDao())
val mediaAccess: MediaAccess = MediaAccessChecker(appContext)
@@ -0,0 +1,31 @@
package com.galo.android.ui.album
import com.galo.android.domain.model.Album
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
/**
* Detects loss of access to an album from its cached value: the album was present
* and then disappeared, which happens when the owner deletes it or the user is
* removed from it.
*
* A cold start without a cached album is not access loss - the album may simply
* not be on the device yet - so the flag only flips after the album was seen.
*/
class AlbumAccessTracker {
private var seen = false
private val lost = MutableStateFlow(false)
val accessLost: StateFlow<Boolean> = lost.asStateFlow()
fun onAlbum(album: Album?) {
if (album != null) {
seen = true
} else if (seen) {
lost.value = true
}
}
}
@@ -82,6 +82,9 @@ fun AlbumScreen(
val viewModel: AlbumViewModel = viewModel(factory = factory)
val state by viewModel.state.collectAsStateWithLifecycle()
// A deleted album (or a lost membership) takes the screen back to the list.
LeaveOnAccessLost(accessLost = state.accessLost, onLeave = onBack)
var panelOpen by remember { mutableStateOf(false) }
var renameOpen by remember { mutableStateOf(false) }
var deleteOpen by remember { mutableStateOf(false) }
@@ -29,6 +29,7 @@ class AlbumViewModel(
val offline: Boolean = false,
val endReached: Boolean = false,
val error: String? = null,
val accessLost: Boolean = false,
)
private data class Status(
@@ -43,6 +44,9 @@ class AlbumViewModel(
private val status = MutableStateFlow(Status())
private val role = MutableStateFlow("")
/** Flags a lost album (deleted or the user removed) so the screen can leave. */
private val access = AlbumAccessTracker()
/** Watermark for `?since=` backfill; advanced as events are applied. */
private var lastSync: String = Instant.now().toString()
@@ -51,7 +55,8 @@ class AlbumViewModel(
container.photoRepository.observePhotos(albumId),
status,
role,
) { album, photos, s, r ->
access.accessLost,
) { album, photos, s, r, accessLost ->
UiState(
album = album,
role = r,
@@ -61,6 +66,7 @@ class AlbumViewModel(
offline = s.offline,
endReached = s.endReached,
error = s.error,
accessLost = accessLost,
)
}.stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), UiState())
@@ -68,6 +74,7 @@ class AlbumViewModel(
viewModelScope.launch { loadAlbum() }
loadNextPage()
observeEvents()
observeAccess()
}
private suspend fun loadAlbum() {
@@ -122,6 +129,13 @@ class AlbumViewModel(
}
}
/** Flags loss of access when the cached album disappears (deleted or revoked). */
private fun observeAccess() {
viewModelScope.launch {
container.albumRepository.observeAlbum(albumId).collect { access.onAlbum(it) }
}
}
/** Live updates while the screen is alive; reconnect backfills missed changes. */
private fun observeEvents() {
viewModelScope.launch {
@@ -0,0 +1,15 @@
package com.galo.android.ui.album
import androidx.compose.runtime.Composable
import androidx.compose.runtime.LaunchedEffect
/**
* Leaves the album screen as soon as access to the album is lost (deleted by its
* owner, or the current user removed from it).
*/
@Composable
fun LeaveOnAccessLost(accessLost: Boolean, onLeave: () -> Unit) {
LaunchedEffect(accessLost) {
if (accessLost) onLeave()
}
}
@@ -3,6 +3,7 @@ package com.galo.android.ui.albums
import androidx.lifecycle.ViewModel
import androidx.lifecycle.viewModelScope
import com.galo.android.di.AppContainer
import com.galo.android.data.repo.LiveAlbumSource
import com.galo.android.domain.model.DeviceAlbum
import com.galo.android.domain.model.SyncMode
import kotlinx.coroutines.flow.MutableStateFlow
@@ -13,7 +14,10 @@ import kotlinx.coroutines.flow.stateIn
import kotlinx.coroutines.flow.update
import kotlinx.coroutines.launch
class AlbumsViewModel(private val container: AppContainer) : ViewModel() {
class AlbumsViewModel(
private val container: AppContainer,
private val live: LiveAlbumSource = container.albumLiveSync,
) : ViewModel() {
data class UiState(
val items: List<AlbumListItem> = emptyList(),
@@ -80,6 +84,9 @@ class AlbumsViewModel(private val container: AppContainer) : ViewModel() {
init {
refresh()
// The personal stream lives as long as the session: it is started once
// here and cancelled with the ViewModel scope on logout.
viewModelScope.launch { live.run() }
}
fun refresh() {