feat(android): двусторонняя синхронизация альбома устройства с сервером

Связанная пара «альбом устройства → серверный альбом» становится миррором:
серверные фото материализуются на устройстве (оригинал в режиме копий, превью/
постер в переносе), состав меняется в обе стороны.

- Признак происхождения (OWN/MIRROR) в индексе и метке провенанса; БД v8.
- Материализация серверных фото (MirrorManager) с датой съёмки и меткой.
- Дедуп, материализация ровно в один бакет при n:1, baseline «только новое»/«всё».
- Двустороннее удаление с учётом прав; для зеркал намерением считается только
  явное удаление в приложении (надгробие), пропавшая локальная копия
  материализуется заново — иначе терялись серверные оригиналы.
- Неудачное чтение серверного списка не считается пустым альбомом.
- Устойчивое выключение синхронизации (syncDisabledAt): восстановление
  происхождения не возвращает связь.
- Очередь загрузки — одно дренирующее задание вместо задания на файл
  (JobScheduler переставал запускать остаток на больших альбомах).
- Дебаунс наблюдателя MediaStore и кэш sha256 по (size, dateModified).

Покрыто unit-тестами и instrumented-тестами (режимы A/B/C и переключения,
двусторонний миррор, объём 100 фото + 20 видео на тест).
This commit is contained in:
iu
2026-10-16 23:19:02 +03:00
parent 19309e2ac0
commit cd326c0ae6
56 changed files with 2766 additions and 88 deletions
@@ -4,6 +4,8 @@ import android.content.Context
import androidx.room.Database
import androidx.room.Room
import androidx.room.RoomDatabase
import androidx.room.migration.Migration
import androidx.sqlite.db.SupportSQLiteDatabase
import com.galo.android.data.db.dao.AlbumDao
import com.galo.android.data.db.dao.PhotoDao
import com.galo.android.data.db.dao.SyncDao
@@ -11,6 +13,7 @@ import com.galo.android.data.db.dao.UploadDao
import com.galo.android.data.db.entity.AlbumEntity
import com.galo.android.data.db.entity.DeviceAlbumMetaEntity
import com.galo.android.data.db.entity.DeviceAssetEntity
import com.galo.android.data.db.entity.DeletedAssetEntity
import com.galo.android.data.db.entity.PhotoEntity
import com.galo.android.data.db.entity.SyncedAlbumEntity
import com.galo.android.data.db.entity.UploadJobEntity
@@ -23,8 +26,9 @@ import com.galo.android.data.db.entity.UploadJobEntity
SyncedAlbumEntity::class,
DeviceAssetEntity::class,
DeviceAlbumMetaEntity::class,
DeletedAssetEntity::class,
],
version = 6,
version = 8,
exportSchema = false,
)abstract class GaloDatabase : RoomDatabase() {
@@ -41,7 +45,43 @@ import com.galo.android.data.db.entity.UploadJobEntity
fun create(context: Context): GaloDatabase =
Room.databaseBuilder(context.applicationContext, GaloDatabase::class.java, NAME)
.addMigrations(MIGRATION_6_7, MIGRATION_7_8)
.fallbackToDestructiveMigration()
.build()
/**
* Adds two-way mirroring state without losing the sync index: the
* provenance origin, the first-pass baseline and the deletion journal.
*/
val MIGRATION_6_7 = object : Migration(6, 7) {
override fun migrate(db: SupportSQLiteDatabase) {
db.execSQL(
"ALTER TABLE synced_albums ADD COLUMN mirrorFromCreatedAt INTEGER NOT NULL DEFAULT 0",
)
db.execSQL(
"ALTER TABLE device_assets ADD COLUMN origin TEXT NOT NULL DEFAULT 'OWN'",
)
db.execSQL(
"CREATE TABLE IF NOT EXISTS deleted_assets (" +
"photoId TEXT NOT NULL, deviceAlbumId TEXT NOT NULL, " +
"status TEXT NOT NULL, createdAt INTEGER NOT NULL, " +
"PRIMARY KEY(photoId))",
)
db.execSQL(
"CREATE INDEX IF NOT EXISTS index_deleted_assets_deviceAlbumId " +
"ON deleted_assets (deviceAlbumId)",
)
}
}
/**
* Records whether the user explicitly turned synchronization off for an
* album, so provenance recovery does not resurrect the link.
*/
val MIGRATION_7_8 = object : Migration(7, 8) {
override fun migrate(db: SupportSQLiteDatabase) {
db.execSQL("ALTER TABLE device_album_meta ADD COLUMN syncDisabledAt INTEGER")
}
}
}
}
@@ -5,6 +5,7 @@ import androidx.room.Query
import androidx.room.Upsert
import com.galo.android.data.db.entity.DeviceAlbumMetaEntity
import com.galo.android.data.db.entity.DeviceAssetEntity
import com.galo.android.data.db.entity.DeletedAssetEntity
import com.galo.android.data.db.entity.SyncedAlbumEntity
import kotlinx.coroutines.flow.Flow
@@ -97,4 +98,24 @@ interface SyncDao {
@Query("DELETE FROM device_assets WHERE deviceAlbumId = :deviceAlbumId")
suspend fun clearAlbum(deviceAlbumId: String)
// --- Deletion journal (tombstones) -------------------------------------
@Query("SELECT * FROM deleted_assets WHERE deviceAlbumId = :deviceAlbumId")
suspend fun getDeletedAssets(deviceAlbumId: String): List<DeletedAssetEntity>
@Query("SELECT * FROM deleted_assets")
suspend fun getAllDeletedAssets(): List<DeletedAssetEntity>
@Query("SELECT * FROM deleted_assets WHERE photoId = :photoId")
suspend fun getDeletedAsset(photoId: String): DeletedAssetEntity?
@Upsert
suspend fun upsertDeletedAsset(asset: DeletedAssetEntity)
@Query("DELETE FROM deleted_assets WHERE photoId = :photoId")
suspend fun deleteDeletedAsset(photoId: String)
@Query("DELETE FROM deleted_assets WHERE deviceAlbumId = :deviceAlbumId")
suspend fun clearDeletedAssets(deviceAlbumId: String)
}
@@ -1,5 +1,6 @@
package com.galo.android.data.db.entity
import androidx.room.ColumnInfo
import androidx.room.Entity
import androidx.room.Index
import androidx.room.PrimaryKey
@@ -12,6 +13,11 @@ data class SyncedAlbumEntity(
val serverAlbumId: String,
val mode: String,
val updatedAt: Long,
/**
* Two-way mirroring baseline: only server photos created at or after this
* epoch (millis) are materialized. 0 means "the whole current album".
*/
@ColumnInfo(defaultValue = "0") val mirrorFromCreatedAt: Long = 0L,
)
/**
@@ -24,6 +30,12 @@ data class DeviceAlbumMetaEntity(
val displayName: String?,
val coverUri: String?,
val updatedAt: Long,
/**
* When the user explicitly turned synchronization off (or unlinked), the
* album is local-only until they enable sync again. Provenance recovery must
* not resurrect the link from the offloaded previews' markers.
*/
val syncDisabledAt: Long? = null,
)
/**
@@ -51,8 +63,29 @@ data class DeviceAssetEntity(
val serverPhotoId: String?,
val uploadedAt: Long?,
val state: String,
/** Uploaded by this client (`OWN`) or materialized from the server (`MIRROR`). */
@ColumnInfo(defaultValue = "OWN") val origin: String = "OWN",
) {
companion object {
fun key(deviceAlbumId: String, mediaStoreId: Long): String = "$deviceAlbumId/$mediaStoreId"
}
}
/**
* A deletion made on the device that still has to be confirmed on the server.
* Keeps a locally removed element from being materialized again before its
* server photo is gone, and makes delete events idempotent.
*/
@Entity(tableName = "deleted_assets", indices = [Index("deviceAlbumId")])
data class DeletedAssetEntity(
@PrimaryKey val photoId: String,
val deviceAlbumId: String,
/** `PENDING` while the server delete is not confirmed; `FAILED` after a refusal. */
val status: String,
val createdAt: Long,
) {
companion object {
const val PENDING = "PENDING"
const val FAILED = "FAILED"
}
}
@@ -3,6 +3,7 @@ package com.galo.android.data.repo
import com.galo.android.data.db.dao.SyncDao
import com.galo.android.data.db.entity.DeviceAlbumMetaEntity
import com.galo.android.data.db.entity.DeviceAssetEntity
import com.galo.android.data.db.entity.DeletedAssetEntity
import com.galo.android.data.db.entity.SyncedAlbumEntity
import com.galo.android.data.media.MediaStoreDeviceSource
import com.galo.android.domain.model.DeviceAlbum
@@ -57,6 +58,7 @@ class SyncRepository(
deviceAlbumName: String,
serverAlbumId: String,
mode: SyncMode,
mirrorFromCreatedAt: Long? = null,
) = dao.upsertSyncedAlbum(
SyncedAlbumEntity(
deviceAlbumId = deviceAlbumId,
@@ -64,10 +66,71 @@ class SyncRepository(
serverAlbumId = serverAlbumId,
mode = mode.name,
updatedAt = System.currentTimeMillis(),
// Keep an existing two-way baseline; a brand-new link starts at "now"
// (only what appears from this point on), which is the safe default.
mirrorFromCreatedAt = mirrorFromCreatedAt
?: dao.getSyncedAlbum(deviceAlbumId)?.mirrorFromCreatedAt
?: System.currentTimeMillis(),
),
)
).also {
// Enabling sync clears any earlier explicit "off", so recovery may work.
markSyncDisabled(deviceAlbumId, false)
}
suspend fun unlink(deviceAlbumId: String) = dao.deleteSyncedAlbum(deviceAlbumId)
/** Sets the two-way mirroring baseline: 0 downloads the whole album. */
suspend fun setMirrorFrom(deviceAlbumId: String, from: Long) {
val current = dao.getSyncedAlbum(deviceAlbumId) ?: return
dao.upsertSyncedAlbum(current.copy(mirrorFromCreatedAt = from))
}
/** True when the user explicitly turned synchronization off for this album. */
suspend fun isSyncDisabled(deviceAlbumId: String): Boolean =
dao.getAlbumMeta(deviceAlbumId)?.syncDisabledAt != null
/**
* Records an explicit in-app deletion of a tracked photo, so the coordinator
* deletes it on the server. A local copy merely going missing (gallery cleanup,
* moved file) is not recorded and is re-materialized instead.
*/
suspend fun markExplicitDelete(deviceAlbumId: String, photoId: String) {
dao.upsertDeletedAsset(
DeletedAssetEntity(
photoId = photoId,
deviceAlbumId = deviceAlbumId,
status = DeletedAssetEntity.PENDING,
createdAt = System.currentTimeMillis(),
),
)
}
/**
* Records that the user turned synchronization off (or unlinked) the album,
* so provenance recovery does not recreate the link from preview markers.
*/
suspend fun markSyncDisabled(deviceAlbumId: String, disabled: Boolean) {
val current = dao.getAlbumMeta(deviceAlbumId)
if (disabled) {
dao.upsertAlbumMeta(
DeviceAlbumMetaEntity(
deviceAlbumId = deviceAlbumId,
displayName = current?.displayName,
coverUri = current?.coverUri,
updatedAt = System.currentTimeMillis(),
syncDisabledAt = System.currentTimeMillis(),
),
)
} else if (current?.syncDisabledAt != null) {
dao.upsertAlbumMeta(current.copy(syncDisabledAt = null, updatedAt = System.currentTimeMillis()))
}
}
suspend fun unlink(deviceAlbumId: String) {
dao.deleteSyncedAlbum(deviceAlbumId)
dao.clearDeletedAssets(deviceAlbumId)
// Remember the explicit "off" so the next provenance recovery does not
// recreate the link from the offloaded previews' markers.
markSyncDisabled(deviceAlbumId, true)
}
// --- Local album metadata (display name, cover) ------------------------
@@ -91,6 +154,7 @@ class SyncRepository(
displayName = name?.trim()?.takeIf { it.isNotEmpty() },
coverUri = current?.coverUri,
updatedAt = System.currentTimeMillis(),
syncDisabledAt = current?.syncDisabledAt,
),
)
}
@@ -104,6 +168,7 @@ class SyncRepository(
displayName = current?.displayName,
coverUri = coverUri,
updatedAt = System.currentTimeMillis(),
syncDisabledAt = current?.syncDisabledAt,
),
)
}
@@ -114,6 +179,7 @@ class SyncRepository(
*/
suspend fun forgetAlbum(deviceAlbumId: String) {
dao.clearAlbum(deviceAlbumId)
dao.clearDeletedAssets(deviceAlbumId)
dao.deleteAlbumMeta(deviceAlbumId)
dao.deleteSyncedAlbum(deviceAlbumId)
}
@@ -1,12 +1,75 @@
package com.galo.android.data.sync
import com.galo.android.data.network.ApiClient
import com.galo.android.data.network.GaloApi
import com.galo.android.data.network.dto.PhotoDto
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.withContext
import okhttp3.Request
import java.time.Instant
import java.time.LocalDateTime
import java.time.ZoneOffset
import java.time.format.DateTimeFormatter
/**
* The object URL to download for a server photo: the original in copy mode,
* otherwise the lightest available preview (for video, a poster frame).
*/
internal fun pickVariant(photo: PhotoDto, full: Boolean): String? =
if (full) {
photo.downloadUrl ?: photo.originalUrl
} else {
photo.previewUrl ?: photo.mediumUrl ?: photo.thumbUrl
}
/** [SyncRemote] backed by the galo REST API. */
class ApiSyncRemote(private val api: GaloApi) : SyncRemote {
class ApiSyncRemote(
private val api: GaloApi,
private val apiClient: ApiClient,
) : SyncRemote {
override suspend fun serverChecksums(serverAlbumId: String): Map<String, String> =
api.albumChecksums(serverAlbumId)
serverPhotos(serverAlbumId).mapNotNull { photo ->
photo.checksum?.let { it to photo.id }
}.toMap()
override suspend fun serverPhotos(serverAlbumId: String): List<ServerPhoto> {
val out = ArrayList<ServerPhoto>()
var cursor: String? = null
while (true) {
val page = api.photos(serverAlbumId, cursor = cursor, limit = 100)
page.photos.forEach { p ->
out += ServerPhoto(
id = p.id,
checksum = p.originalChecksum?.ifBlank { null },
displayName = p.originalFilename,
mimeType = p.mimeType,
isVideo = p.kind == "video",
takenAtMillis = parseTakenAt(p.takenAt),
createdAtMillis = parseInstant(p.createdAt) ?: 0L,
width = p.width,
height = p.height,
size = p.size,
)
}
val next = page.nextCursor
if (next.isNullOrBlank()) break
cursor = next
}
return out
}
override suspend fun fetchPhoto(photo: ServerPhoto, full: Boolean): ByteArray? =
withContext(Dispatchers.IO) {
val dto = runCatching { api.photo(photo.id) }.getOrNull() ?: return@withContext null
val raw = pickVariant(dto, full)
val url = apiClient.objectUrls.resolve(raw) ?: return@withContext null
runCatching {
apiClient.client.newCall(Request.Builder().url(url).build()).execute().use { response ->
if (!response.isSuccessful) null else response.body?.bytes()
}
}.getOrNull()
}
override suspend fun renamePhoto(photoId: String, newName: String) {
api.renamePhoto(photoId, newName)
@@ -19,4 +82,24 @@ class ApiSyncRemote(private val api: GaloApi) : SyncRemote {
override suspend fun movePhotos(photoIds: List<String>, serverAlbumId: String) {
api.movePhotos(photoIds, serverAlbumId)
}
private companion object {
val EXIF_FORMAT: DateTimeFormatter = DateTimeFormatter.ofPattern("yyyy:MM:dd HH:mm:ss")
val PLAIN_FORMAT: DateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")
/** ISO instant, EXIF or plain datetime; null when unparseable. */
fun parseInstant(raw: String?): Long? {
val value = raw?.trim()?.ifBlank { null } ?: return null
runCatching { return Instant.parse(value).toEpochMilli() }
runCatching {
return LocalDateTime.parse(value, EXIF_FORMAT).toInstant(ZoneOffset.UTC).toEpochMilli()
}
runCatching {
return LocalDateTime.parse(value, PLAIN_FORMAT).toInstant(ZoneOffset.UTC).toEpochMilli()
}
return null
}
fun parseTakenAt(raw: String?): Long? = parseInstant(raw)
}
}
@@ -10,23 +10,34 @@ import android.provider.MediaStore
/**
* Watches MediaStore while the app process is alive and notifies on changes so
* synchronization can run promptly in the foreground. Background changes are
* covered by the periodic worker.
* covered by the periodic worker. Bursts of changes (for example materializing or
* offloading a whole album) are coalesced into a single notification after
* [debounceMs], so one file write does not trigger a full synchronization pass.
*/
class DeviceMediaObserver(
private val context: Context,
private val onChanged: () -> Unit,
private val debounceMs: Long = 1_500,
) {
private val observer = object : ContentObserver(Handler(Looper.getMainLooper())) {
private val handler = Handler(Looper.getMainLooper())
private val notify = Runnable { onChanged() }
private val observer = object : ContentObserver(handler) {
override fun onChange(selfChange: Boolean) {
onChanged()
schedule()
}
override fun onChange(selfChange: Boolean, uri: Uri?) {
onChanged()
schedule()
}
}
private fun schedule() {
handler.removeCallbacks(notify)
handler.postDelayed(notify, debounceMs)
}
fun start() {
runCatching {
context.contentResolver.registerContentObserver(
@@ -44,5 +55,6 @@ class DeviceMediaObserver(
fun stop() {
runCatching { context.contentResolver.unregisterContentObserver(observer) }
handler.removeCallbacks(notify)
}
}
@@ -8,6 +8,7 @@ import com.galo.android.data.media.MediaStoreDeviceSource
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.withContext
import java.io.InputStream
import java.util.concurrent.ConcurrentHashMap
/** Supplies the device's current content to the synchronization engine. */
interface LocalAssetProvider {
@@ -34,7 +35,7 @@ class DeviceLocalAssetProvider(
displayName = media.displayName,
mimeType = media.mimeType,
kind = media.kind,
contentHash = hash(media.uri),
contentHash = hashCached(media.uri, media.size, media.dateModified),
uri = media.uri,
)
}
@@ -46,6 +47,23 @@ class DeviceLocalAssetProvider(
override fun openStream(uri: String): InputStream? =
runCatching { context.contentResolver.openInputStream(Uri.parse(uri)) }.getOrNull()
private data class CachedHash(val size: Long, val dateModified: Long, val hash: String?)
// Hashing a large album every pass is expensive; MediaStore updates
// DATE_MODIFIED on any write, so an unchanged (size, dateModified) pair lets
// us reuse the previous sha256 instead of re-reading the whole file.
private val hashCache = ConcurrentHashMap<String, CachedHash>()
private fun hashCached(uri: String, size: Long, dateModified: Long): String? {
val cached = hashCache[uri]
if (cached != null && cached.size == size && cached.dateModified == dateModified) {
return cached.hash
}
val hash = hash(uri)
hashCache[uri] = CachedHash(size, dateModified, hash)
return hash
}
private fun hash(uri: String): String? =
runCatching { openStream(uri)?.use { ContentHash.sha256(it) } }.getOrNull()
}
@@ -0,0 +1,174 @@
package com.galo.android.data.sync
import android.content.ContentValues
import android.content.Context
import android.net.Uri
import android.provider.MediaStore
import com.galo.android.data.media.MediaStoreDeviceSource
import com.galo.android.domain.model.SyncMode
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import kotlinx.coroutines.withContext
import java.io.File
/** Outcome of materializing one server photo on the device. */
sealed interface MirrorResult {
data class Done(val mediaStoreId: Long, val uri: String) : MirrorResult
/** Nothing could be downloaded or written. */
data object Failed : MirrorResult
}
/** Writes server photos into a device album's MediaStore bucket, and removes them. */
interface AssetMirror {
suspend fun materialize(
deviceAlbumId: String,
serverAlbumId: String,
mode: SyncMode,
photo: ServerPhoto,
): MirrorResult
/**
* Removes a materialized file from the gallery when its server photo is gone.
* @return true when the media is gone; false when the app does not own it.
*/
suspend fun remove(uri: String): Boolean
}
/**
* Downloads a server photo and inserts it into the album's MediaStore bucket:
* the original in copy mode, a light preview (poster for video) in move mode.
* The inserted file carries an `origin=MIRROR` provenance marker and the capture
* date, and belongs to the app, so a later server deletion can remove it without
* a system consent dialog.
*/
class MirrorManager(
private val context: Context,
private val remote: SyncRemote,
private val deviceSource: MediaStoreDeviceSource,
) : AssetMirror {
private val mutex = Mutex()
override suspend fun remove(uri: String): Boolean = withContext(Dispatchers.IO) {
runCatching {
context.contentResolver.delete(Uri.parse(uri), null, null) > 0
}.getOrDefault(false)
}
override suspend fun materialize(
deviceAlbumId: String,
serverAlbumId: String,
mode: SyncMode,
photo: ServerPhoto,
): MirrorResult = withContext(Dispatchers.IO) {
mutex.withLock {
val full = mode == SyncMode.COPY
val bytes = runCatching { remote.fetchPhoto(photo, full) }.getOrNull()
if (bytes == null || bytes.isEmpty()) {
android.util.Log.w(TAG, "fetch failed ${photo.id} video=${photo.isVideo} full=$full")
return@withLock MirrorResult.Failed
}
runCatching { insert(deviceAlbumId, serverAlbumId, mode, photo, bytes) }
.onFailure { android.util.Log.w(TAG, "insert failed ${photo.id}: ${it.message}") }
.getOrDefault(MirrorResult.Failed)
}
}
private fun insert(
deviceAlbumId: String,
serverAlbumId: String,
mode: SyncMode,
photo: ServerPhoto,
bytes: ByteArray,
): MirrorResult {
val asVideo = photo.isVideo && mode == SyncMode.COPY
val relativePath = relativePathOf(deviceAlbumId)
val mime = when {
asVideo -> photo.mimeType?.takeIf { it.startsWith("video/") } ?: "video/mp4"
photo.isVideo -> "image/jpeg"
else -> photo.mimeType?.takeIf { it.startsWith("image/") } ?: "image/jpeg"
}
val displayName = displayNameOf(photo, asVideo)
val collection = if (asVideo) {
MediaStore.Video.Media.EXTERNAL_CONTENT_URI
} else {
MediaStore.Images.Media.EXTERNAL_CONTENT_URI
}
val values = ContentValues().apply {
put(MediaStore.MediaColumns.DISPLAY_NAME, displayName)
put(MediaStore.MediaColumns.MIME_TYPE, mime)
put(MediaStore.MediaColumns.RELATIVE_PATH, relativePath)
put(MediaStore.MediaColumns.IS_PENDING, 1)
photo.takenAtMillis?.let { put(MediaStore.MediaColumns.DATE_TAKEN, it) }
}
val inserted = context.contentResolver.insert(collection, values) ?: return MirrorResult.Failed
val temp = tempFile(displayName)
try {
temp.writeBytes(bytes)
runCatching {
ProvenanceStamp.write(
temp,
Provenance(
serverAlbumId = serverAlbumId,
serverPhotoId = photo.id,
originalHash = photo.checksum,
deviceAlbumId = deviceAlbumId,
mode = mode.name,
assetKey = null,
origin = "MIRROR",
),
)
}
context.contentResolver.openOutputStream(inserted)?.use { out ->
temp.inputStream().use { it.copyTo(out) }
} ?: return MirrorResult.Failed
values.clear()
values.put(MediaStore.MediaColumns.IS_PENDING, 0)
context.contentResolver.update(inserted, values, null, null)
// MediaProvider may recompute DATE_TAKEN when the pending flag clears,
// so set the capture date again in a separate update.
photo.takenAtMillis?.let { taken ->
val date = android.content.ContentValues().apply {
put(MediaStore.MediaColumns.DATE_TAKEN, taken)
}
context.contentResolver.update(inserted, date, null, null)
}
} finally {
temp.delete()
}
return MirrorResult.Done(
mediaStoreId = android.content.ContentUris.parseId(inserted),
uri = inserted.toString(),
)
}
/** Writes the poster/original next to the album's existing files. */
private fun relativePathOf(deviceAlbumId: String): String =
deviceSource.items(deviceAlbumId)
.firstOrNull { it.relativePath.isNotBlank() }
?.relativePath
?: "DCIM/"
private fun displayNameOf(photo: ServerPhoto, asVideo: Boolean): String {
val base = photo.displayName.ifBlank { photo.id }
return when {
asVideo -> base
photo.isVideo -> base.substringBeforeLast('.') + "-preview.jpg"
else -> if (base.contains('.')) base else "$base.jpg"
}
}
private fun tempFile(name: String): File {
val dir = File(context.cacheDir, "mirror").apply { mkdirs() }
val suffix = name.substringAfterLast('.', "bin")
return File(dir, "mirror-${System.nanoTime()}.$suffix")
}
private companion object {
const val TAG = "MirrorManager"
}
}
@@ -200,6 +200,7 @@ class OffloadManager(
// this back to rebuild the link after a reinstall.
mode = syncMode ?: SyncMode.MOVE.name,
assetKey = asset.assetKey,
origin = asset.origin,
)
private suspend fun markPreview(asset: DeviceAssetEntity) {
@@ -16,6 +16,8 @@ data class Provenance(
val deviceAlbumId: String?,
val mode: String?,
val assetKey: String?,
/** `OWN` when this client uploaded the element, `MIRROR` when downloaded. */
val origin: String? = null,
)
/**
@@ -49,6 +51,7 @@ object ProvenanceStamp {
append('|').append("device=").append(p.deviceAlbumId ?: "")
append('|').append("mode=").append(p.mode ?: "")
append('|').append("asset=").append(p.assetKey ?: "")
append('|').append("origin=").append(p.origin ?: "")
}
fun decode(raw: String?): Provenance? {
@@ -66,6 +69,7 @@ object ProvenanceStamp {
deviceAlbumId = map["device"]?.ifBlank { null },
mode = map["mode"]?.ifBlank { null },
assetKey = map["asset"]?.ifBlank { null },
origin = map["origin"]?.ifBlank { null },
)
}
@@ -1,8 +1,10 @@
package com.galo.android.data.sync
import com.galo.android.data.db.dao.SyncDao
import com.galo.android.data.db.entity.DeletedAssetEntity
import com.galo.android.data.db.entity.DeviceAssetEntity
import com.galo.android.data.db.entity.SyncedAlbumEntity
import com.galo.android.domain.model.AssetOrigin
import com.galo.android.domain.model.LocalAssetState
import com.galo.android.domain.model.SyncMode
import com.galo.android.domain.sync.SyncPolicy
@@ -19,9 +21,11 @@ sealed interface RetargetResult {
}
/**
* Reconciles a device album with its server album. Mode A (DISABLED) is local
* only and performs no server requests; the other modes upload missing content,
* dedup by checksum, propagate renames and update the local index.
* Reconciles a device album with its server album as a two-way mirror. Mode A
* (DISABLED) is local only and performs no server requests; the other modes
* upload missing content and materialize server photos (originals in copy mode,
* previews in move mode), dedup by checksum, propagate renames and deletions in
* both directions, and update the local index.
*/
class SyncCoordinator(
private val dao: SyncDao,
@@ -29,6 +33,7 @@ class SyncCoordinator(
private val remote: SyncRemote,
private val uploadQueue: SyncUploadQueue,
private val offloader: AssetOffloader,
private val mirror: AssetMirror,
) {
/** Serializes reconciles so rapid triggers never double-upload or double-offload. */
@@ -45,19 +50,42 @@ class SyncCoordinator(
// A mode can be chosen before a server album; nothing to sync until then.
if (link.serverAlbumId.isBlank()) return false
val serverChecksums = remote.serverChecksums(link.serverAlbumId)
val livePhotoIds = serverChecksums.values.toSet()
// If the album could not be read, do NOT treat it as empty: an empty list
// would make the engine think every server photo is gone and delete the
// local mirrors. Skip the pass instead.
val serverPhotos = runCatching { remote.serverPhotos(link.serverAlbumId) }.getOrNull()
?: return false
val serverChecksums = serverPhotos.mapNotNull { photo ->
photo.checksum?.let { it to photo.id }
}.toMap()
val livePhotoIds = serverPhotos.mapTo(HashSet()) { it.id }
// A tombstone whose photo is gone from the server is confirmed: stop
// hiding it. Anything still pending keeps the element from coming back.
clearConfirmedTombstones(deviceAlbumId, livePhotoIds)
val assets = local.assets(deviceAlbumId)
val localKeys = assets.mapTo(HashSet()) { it.assetKey }
val tombstones = dao.getDeletedAssets(deviceAlbumId).mapTo(HashSet()) { it.photoId }
// Reflect lost/returned server copies before planning, so an offloaded
// preview whose photo is gone is seen as ORPHANED and handled as such.
val index = reflectLostCopies(dao.getAssets(deviceAlbumId), livePhotoIds).map {
IndexedAsset(it.assetKey, it.serverPhotoId, it.contentHash, it.displayName, it.state)
val reflected = reflectLostCopies(dao.getAssets(deviceAlbumId), livePhotoIds)
// Materialized elements are reconciled against the server set. Rows with an
// explicit deletion are left for the planner to delete on the server.
reconcileMirrors(reflected, livePhotoIds, localKeys, tombstones)
val indexEntities = dao.getAssets(deviceAlbumId)
val index = indexEntities.map {
IndexedAsset(it.assetKey, it.serverPhotoId, it.contentHash, it.displayName, it.state, it.origin)
}
val hashesOutside = local.hashesOutside(deviceAlbumId)
val actions = SyncPlanner.plan(assets, index, serverChecksums, hashesOutside)
val actions = SyncPlanner.plan(assets, index, serverChecksums, hashesOutside, tombstones)
for (action in actions) apply(deviceAlbumId, link.serverAlbumId, mode, action, livePhotoIds)
// Two-way: bring the target album's photos onto the device.
materializeMissing(deviceAlbumId, link, mode, serverPhotos, assets, indexEntities)
// Mode C: assets already on the server (e.g. after switching from copies)
// are replaced with their offloaded versions, right away, but only while
// their copy is confirmed in the target album.
@@ -79,7 +107,8 @@ class SyncCoordinator(
newServerAlbumId: String,
mode: SyncMode,
): RetargetResult = mutex.withLock {
val previous = dao.getSyncedAlbum(deviceAlbumId)
val prior = dao.getSyncedAlbum(deviceAlbumId)
val previous = prior
?.serverAlbumId
?.takeIf { it.isNotBlank() }
val recordedPhotoIds = dao.getAssets(deviceAlbumId)
@@ -121,6 +150,7 @@ class SyncCoordinator(
serverAlbumId = newServerAlbumId,
mode = mode.name,
updatedAt = System.currentTimeMillis(),
mirrorFromCreatedAt = prior?.mirrorFromCreatedAt ?: System.currentTimeMillis(),
),
)
reconcileLocked(deviceAlbumId)
@@ -143,12 +173,14 @@ class SyncCoordinator(
/**
* Reflects the loss or return of the server copy behind an offloaded preview:
* a preview whose server photo is gone becomes ORPHANED, and one whose photo is
* back becomes PREVIEW again. Returns the updated index rows for planning.
* back becomes PREVIEW again. Materialized elements are handled separately.
* Returns the updated index rows for planning.
*/
private suspend fun reflectLostCopies(
entities: List<DeviceAssetEntity>,
livePhotoIds: Set<String>,
): List<DeviceAssetEntity> = entities.map { entity ->
if (entity.origin == AssetOrigin.MIRROR.name) return@map entity
val photoId = entity.serverPhotoId ?: return@map entity
val live = photoId in livePhotoIds
when {
@@ -166,6 +198,128 @@ class SyncCoordinator(
}
}
/**
* Reconciles materialized (server-origin) elements:
* - if the server photo is gone, the local mirror is removed (two-way deletion);
* - if the server photo is still there but the local copy is missing, the stale
* index row is dropped so the photo is materialized again — the client must
* never delete a server photo just because it lost its own copy of it.
*/
private suspend fun reconcileMirrors(
entities: List<DeviceAssetEntity>,
livePhotoIds: Set<String>,
localKeys: Set<String>,
tombstones: Set<String>,
) {
for (entity in entities) {
if (entity.origin != AssetOrigin.MIRROR.name) continue
val photoId = entity.serverPhotoId ?: continue
// An explicit in-app deletion is handled by the planner (delete on the
// server); do not drop the row here or the deletion would be lost.
if (photoId in tombstones) continue
if (photoId in livePhotoIds) {
if (entity.assetKey !in localKeys) dao.deleteAsset(entity.assetKey)
continue
}
val removed = runCatching { mirror.remove(entity.uri) }.getOrDefault(false)
if (removed) {
dao.deleteAsset(entity.assetKey)
} else {
dao.markState(entity.assetKey, LocalAssetState.ORPHANED.name)
}
}
}
/** Clears tombstones whose server photo is gone (delete confirmed). */
private suspend fun clearConfirmedTombstones(deviceAlbumId: String, livePhotoIds: Set<String>) {
for (tombstone in dao.getDeletedAssets(deviceAlbumId)) {
if (tombstone.photoId !in livePhotoIds) dao.deleteDeletedAsset(tombstone.photoId)
}
}
/**
* Materializes the server photos that are not represented locally yet, are
* not pending deletion, and whose content is not already on the device. Only
* photos at or after the link's baseline are considered.
*/
private suspend fun materializeMissing(
deviceAlbumId: String,
link: SyncedAlbumEntity,
mode: SyncMode,
serverPhotos: List<ServerPhoto>,
assets: List<LocalAsset>,
indexEntities: List<DeviceAssetEntity>,
) {
// Several device albums may point at one server album (n:1). Materialize
// into exactly one of them (a stable choice) so the gallery never gets a
// duplicate from the same server photo.
if (!isMirrorTarget(deviceAlbumId, link.serverAlbumId)) return
val represented = indexEntities.mapNotNull { it.serverPhotoId }.toSet()
val deviceHashes = HashSet<String>()
assets.forEach { it.contentHash?.let(deviceHashes::add) }
indexEntities.forEach { it.contentHash?.let(deviceHashes::add) }
val tombstones = dao.getDeletedAssets(deviceAlbumId).mapTo(HashSet()) { it.photoId }
val from = link.mirrorFromCreatedAt
val eligible = serverPhotos.filter {
from == 0L || it.createdAtMillis == 0L || it.createdAtMillis >= from
}
val toDownload = SyncPlanner.mirrorPlan(eligible, represented, deviceHashes, tombstones)
if (toDownload.isNotEmpty()) {
android.util.Log.i(
TAG,
"materialize plan=${toDownload.size} server=${serverPhotos.size} represented=${represented.size}",
)
}
for (photo in toDownload) {
val result = runCatching {
mirror.materialize(deviceAlbumId, link.serverAlbumId, mode, photo)
}.getOrNull() ?: continue
if (result !is MirrorResult.Done) continue
recordMirror(deviceAlbumId, mode, photo, result)
}
}
/**
* Among all device albums linked to [serverAlbumId], the first by id owns
* materialization. Stable, so a photo is never written into two buckets.
*/
private suspend fun isMirrorTarget(deviceAlbumId: String, serverAlbumId: String): Boolean {
val links = dao.getSyncedAlbumsForServer(serverAlbumId)
if (links.size <= 1) return true
return links.minByOrNull { it.deviceAlbumId }?.deviceAlbumId == deviceAlbumId
}
/** Records a materialized server photo in the index with `origin=MIRROR`. */
private suspend fun recordMirror( deviceAlbumId: String,
mode: SyncMode,
photo: ServerPhoto,
result: MirrorResult.Done,
) {
val state = if (mode == SyncMode.MOVE) LocalAssetState.PREVIEW else LocalAssetState.UPLOADED
dao.upsertAssets(
listOf(
DeviceAssetEntity(
assetKey = DeviceAssetEntity.key(deviceAlbumId, result.mediaStoreId),
deviceAlbumId = deviceAlbumId,
mediaStoreId = result.mediaStoreId,
uri = result.uri,
displayName = photo.displayName,
mimeType = photo.mimeType,
size = photo.size,
dateModified = 0L,
contentHash = photo.checksum,
serverPhotoId = photo.id,
uploadedAt = null,
state = state.name,
origin = AssetOrigin.MIRROR.name,
),
),
)
}
/**
* Offloads assets that are already on the server but still have their original
* locally (mode C). Retries assets left as "awaiting free" from a previous pass,
@@ -224,14 +378,27 @@ class SyncCoordinator(
}
is SyncAction.RenameOnServer -> {
remote.renamePhoto(action.photoId, action.newName)
runCatching { remote.renamePhoto(action.photoId, action.newName) }
dao.updateDisplayNameByServerPhoto(action.photoId, action.newName)
}
is SyncAction.DeleteOnServer -> {
// A photo already gone from the server is only reflected locally:
// the device original is never touched and nothing is re-deleted.
if (action.photoId in livePhotoIds) remote.deletePhoto(action.photoId)
if (action.photoId in livePhotoIds) {
val ok = runCatching { remote.deletePhoto(action.photoId) }.isSuccess
// Record the deletion so the element is not materialized again
// before the server confirms it. A refusal is kept as FAILED so
// it is not retried forever and stays visible as a problem.
dao.upsertDeletedAsset(
DeletedAssetEntity(
photoId = action.photoId,
deviceAlbumId = deviceAlbumId,
status = if (ok) DeletedAssetEntity.PENDING else DeletedAssetEntity.FAILED,
createdAt = System.currentTimeMillis(),
),
)
}
dao.deleteAsset(action.assetKey)
}
}
@@ -263,4 +430,8 @@ class SyncCoordinator(
),
)
}
private companion object {
const val TAG = "SyncCoordinator"
}
}
@@ -1,5 +1,6 @@
package com.galo.android.data.sync
import com.galo.android.domain.model.AssetOrigin
import com.galo.android.domain.model.LocalAssetState
import com.galo.android.domain.model.MediaKind
@@ -21,6 +22,8 @@ data class IndexedAsset(
val contentHash: String?,
val displayName: String,
val state: String = "",
/** `OWN` (uploaded by this client) or `MIRROR` (materialized from the server). */
val origin: String = "OWN",
)
/** A change to apply to the server album. */
@@ -54,6 +57,7 @@ object SyncPlanner {
index: List<IndexedAsset>,
serverChecksums: Map<String, String>,
hashesPresentElsewhere: Set<String>,
explicitlyDeleted: Set<String> = emptySet(),
): List<SyncAction> {
val actions = mutableListOf<SyncAction>()
val livePhotoIds = serverChecksums.values.toSet()
@@ -103,6 +107,10 @@ object SyncPlanner {
for (indexed in index) {
val photoId = indexed.serverPhotoId ?: continue
if (indexed.assetKey in localKeys) continue
// A materialized (server-origin) photo is deleted on the server only on
// an explicit in-app deletion (a tombstone). A merely missing local copy
// is not an intent to delete: the coordinator recreates it instead.
if (indexed.origin == AssetOrigin.MIRROR.name && photoId !in explicitlyDeleted) continue
val hash = indexed.contentHash
if (hash != null && hash in hashesPresentElsewhere) continue
// The remote delete only happens while this photo is in the target
@@ -112,4 +120,20 @@ object SyncPlanner {
return actions
}
/**
* Server photos that must be materialized on the device: every photo of the
* target album that is not already represented locally, not pending deletion,
* and whose content is not already present on the device (dedup).
*/
fun mirrorPlan(
serverPhotos: List<ServerPhoto>,
representedPhotoIds: Set<String>,
deviceHashes: Set<String>,
tombstonedPhotoIds: Set<String>,
): List<ServerPhoto> = serverPhotos.filter { photo ->
photo.id !in representedPhotoIds &&
photo.id !in tombstonedPhotoIds &&
(photo.checksum == null || photo.checksum !in deviceHashes)
}
}
@@ -6,6 +6,7 @@ import com.galo.android.data.db.dao.SyncDao
import com.galo.android.data.db.entity.DeviceAssetEntity
import com.galo.android.data.db.entity.SyncedAlbumEntity
import com.galo.android.data.media.MediaStoreDeviceSource
import com.galo.android.domain.model.AssetOrigin
import com.galo.android.domain.model.LocalAssetState
import com.galo.android.domain.model.SyncMode
import kotlinx.coroutines.Dispatchers
@@ -25,6 +26,10 @@ class SyncRecovery(
/** @return how many assets were recovered from their markers. */
suspend fun recover(deviceAlbumId: String): Int = withContext(Dispatchers.IO) {
// The user explicitly turned synchronization off for this album: the
// preview markers must not resurrect the link they unlinked.
if (dao.getAlbumMeta(deviceAlbumId)?.syncDisabledAt != null) return@withContext 0
var recovered = 0
for (media in deviceSource.items(deviceAlbumId)) {
val marker = ProvenanceStamp.read(context, Uri.parse(media.uri)) ?: continue
@@ -45,6 +50,7 @@ class SyncRecovery(
serverPhotoId = photoId,
uploadedAt = null,
state = LocalAssetState.PREVIEW.name,
origin = AssetOrigin.from(marker.origin).name,
),
),
)
@@ -1,10 +1,39 @@
package com.galo.android.data.sync
/**
* A server photo of the target album as the synchronization engine sees it.
* Enough to decide whether it must be materialized on the device and to fetch
* the right variant.
*/
data class ServerPhoto(
val id: String,
val checksum: String?,
val displayName: String,
val mimeType: String?,
val isVideo: Boolean,
/** Capture time in epoch millis, when known. */
val takenAtMillis: Long?,
/** Server creation time in epoch millis; 0 when unknown. */
val createdAtMillis: Long,
val width: Int,
val height: Int,
val size: Long,
)
/** Server-side operations the synchronization engine needs. */
interface SyncRemote {
/** Control sums of the server photos in an album, keyed by checksum. */
suspend fun serverChecksums(serverAlbumId: String): Map<String, String>
/** The album's photos with the metadata needed to materialize them. */
suspend fun serverPhotos(serverAlbumId: String): List<ServerPhoto>
/**
* Downloads a server photo: the original when [full] is true, otherwise a
* light preview (for video, a poster frame).
*/
suspend fun fetchPhoto(photo: ServerPhoto, full: Boolean): ByteArray?
suspend fun renamePhoto(photoId: String, newName: String)
suspend fun deletePhoto(photoId: String)
@@ -9,7 +9,6 @@ import androidx.work.ExistingWorkPolicy
import androidx.work.NetworkType
import androidx.work.OneTimeWorkRequestBuilder
import androidx.work.WorkManager
import androidx.work.workDataOf
import com.galo.android.data.db.dao.UploadDao
import com.galo.android.data.db.entity.UploadJobEntity
import com.galo.android.domain.model.SyncUpload
@@ -71,9 +70,9 @@ class UploadRepository(
convertedPath = null,
)
dao.upsert(job)
UploadScheduler.enqueue(context, job.id, wifiOnly)
queued++
}
if (queued > 0) UploadScheduler.enqueueDrain(context)
return queued
}
@@ -112,17 +111,22 @@ class UploadRepository(
syncMode = mode,
)
dao.upsert(job)
UploadScheduler.enqueue(context, job.id, wifiOnly)
UploadScheduler.enqueueDrain(context)
}
/** Re-schedules unfinished work after a restart and clears stale RUNNING status. */
suspend fun requeuePending() {
dao.resetRunning()
dao.getPending().forEach { job ->
UploadScheduler.enqueue(context, job.id, job.wifiOnly)
}
if (dao.getPending().isNotEmpty()) UploadScheduler.enqueueDrain(context)
}
/** Pending entries in queue order; the draining worker processes them. */
suspend fun pendingJobs(): List<UploadJobEntity> = dao.getPending()
suspend fun hasPending(): Boolean = dao.getPending().isNotEmpty()
suspend fun resetRunning() = dao.resetRunning()
suspend fun getById(id: String): UploadJobEntity? = dao.getById(id)
/** True when a sync upload for this asset is already queued or running. */
@@ -161,22 +165,20 @@ class UploadRepository(
object UploadScheduler {
const val KEY_JOB_ID = "job_id"
/** One draining job processes the whole queue, instead of one job per file. */
const val UNIQUE_DRAIN = "galo-upload-drain"
fun enqueue(context: Context, jobId: String, wifiOnly: Boolean) {
fun enqueueDrain(context: Context) {
val constraints = Constraints.Builder()
.setRequiredNetworkType(
if (wifiOnly) NetworkType.UNMETERED else NetworkType.CONNECTED,
)
.setRequiredNetworkType(NetworkType.CONNECTED)
.build()
val request = OneTimeWorkRequestBuilder<UploadWorker>()
.setInputData(workDataOf(KEY_JOB_ID to jobId))
.setConstraints(constraints)
.setBackoffCriteria(BackoffPolicy.EXPONENTIAL, 15, TimeUnit.SECONDS)
.addTag(TAG_UPLOAD)
.build()
WorkManager.getInstance(context)
.enqueueUniqueWork(jobId, ExistingWorkPolicy.KEEP, request)
.enqueueUniqueWork(UNIQUE_DRAIN, ExistingWorkPolicy.KEEP, request)
}
const val TAG_UPLOAD = "galo-upload"
@@ -2,6 +2,8 @@ package com.galo.android.data.upload
import android.content.Context
import android.content.pm.ServiceInfo
import android.net.ConnectivityManager
import android.net.NetworkCapabilities
import android.net.Uri
import androidx.core.app.NotificationCompat
import androidx.work.CoroutineWorker
@@ -23,58 +25,89 @@ import java.io.File
import java.io.IOException
/**
* Uploads a single queue entry as one multipart POST. Using a foreground
* notification while running lets large/video uploads continue when the app is
* backgrounded; transient failures retry with WorkManager's exponential backoff
* up to [MAX_ATTEMPTS].
* Drains the persistent upload queue: one WorkManager job processes every pending
* entry sequentially (one file per HTTP request, per the server's fail-fast,
* size-capped uploads). A single draining job avoids the JobScheduler limits and
* foreground-service churn of one job per file, so a large album uploads reliably.
*/
class UploadWorker(
appContext: Context,
params: WorkerParameters,
) : CoroutineWorker(appContext, params) {
private enum class Outcome { DONE, RETRY }
override suspend fun doWork(): Result {
val app = applicationContext as? GaloApplication ?: return Result.failure()
val container = app.container
val repository = container.uploadRepository
val jobId = inputData.getString(UploadScheduler.KEY_JOB_ID) ?: return Result.failure()
val job = repository.getById(jobId) ?: return Result.success()
if (job.status == UploadStatus.SUCCESS.name) return Result.success()
// A previous run may have been killed mid-file; those entries are re-queued.
repository.resetRunning()
val pending = repository.pendingJobs()
if (pending.isEmpty()) return Result.success()
val attempt = runAttemptCount
var retry = false
for (job in pending) {
// A job limited to Wi-Fi is skipped while on a metered network.
if (job.wifiOnly && !isUnmetered()) {
retry = true
continue
}
if (process(container, repository, job) == Outcome.RETRY) retry = true
}
// Entries added while this pass ran (or transient failures) need another pass.
if (repository.hasPending()) retry = true
return if (retry) Result.retry() else Result.success()
}
private suspend fun process(
container: AppContainer,
repository: UploadRepository,
job: UploadJobEntity,
): Outcome {
val jobId = job.id
val attempt = job.attempts
repository.updateState(jobId, UploadStatus.RUNNING, job.progress, null, attempt, job.convertedPath)
var tempPath = job.convertedPath
var lastPercent = -1
return try {
setForeground(foregroundInfo(jobId, job.filename, 0))
setForegroundBestEffort(job.filename, 0)
val uri = Uri.parse(job.uri)
val (filename, body) = prepareBody(uri, job.filename, job.mimeType, job.size) { percent ->
onProgress(repository, jobId, job.filename, percent, attempt, tempPath)
}.also { if (it.tempPath != null) tempPath = it.tempPath }
val prepared = prepareBody(uri, job.filename, job.mimeType, job.size) { percent ->
if (percent != lastPercent) {
lastPercent = percent
runBlocking(Dispatchers.IO) {
repository.updateState(jobId, UploadStatus.RUNNING, percent, null, attempt, tempPath)
setForegroundBestEffort(job.filename, percent)
}
}
}
if (prepared.tempPath != null) tempPath = prepared.tempPath
val photo = container.api.upload(job.albumId, filename, body)
val photo = container.api.upload(job.albumId, prepared.filename, prepared.body)
container.photoRepository.upsert(photo)
recordProvenance(container, job, photo.id)
repository.updateState(jobId, UploadStatus.SUCCESS, 100, null, attempt, tempPath)
tempPath?.let { runCatching { File(it).delete() } }
Result.success()
Outcome.DONE
} catch (e: Exception) {
val message = describe(e)
if (attempt + 1 >= MAX_ATTEMPTS) {
repository.updateState(jobId, UploadStatus.ERROR, job.progress, message, attempt, tempPath)
Result.failure()
Outcome.DONE
} else {
repository.updateState(jobId, UploadStatus.QUEUED, job.progress, message, attempt, tempPath)
Result.retry()
repository.updateState(jobId, UploadStatus.QUEUED, job.progress, message, attempt + 1, tempPath)
Outcome.RETRY
}
}
}
/**
* For a job created by the device-album sync, records the created server photo
* as the provenance of the source asset so later passes can dedup and delete.
* Records the created server photo as provenance of the source asset (sync
* jobs) and, in mode C, offloads the original right after the upload.
*/
private suspend fun recordProvenance(container: AppContainer, job: UploadJobEntity, photoId: String) {
val deviceAlbumId = job.deviceAlbumId ?: return
@@ -92,12 +125,8 @@ class UploadWorker(
serverPhotoId = photoId,
state = state,
)
// Mode C: as soon as the object is uploaded, replace the local original
// with the offloaded (preview) version. Media the app does not own needs
// the OS write consent, which is requested from the album screen.
if (mode == SyncMode.MOVE) {
container.syncRepository.getAsset(assetKey)?.let { asset ->
// The upload just succeeded, so the copy is live in the target.
runCatching { container.offloadManager.offload(asset, setOf(photoId)) }
}
}
@@ -162,23 +191,16 @@ class UploadWorker(
return -1L
}
private fun onProgress(
repository: UploadRepository,
jobId: String,
filename: String,
percent: Int,
attempt: Int,
tempPath: String?,
) {
if (percent == lastPercent) return
lastPercent = percent
runBlocking(Dispatchers.IO) {
repository.updateState(jobId, UploadStatus.RUNNING, percent, null, attempt, tempPath)
setForeground(foregroundInfo(jobId, filename, percent))
}
/**
* Foreground is best-effort: the OS may refuse to start a foreground service
* from the background. The upload still runs; it just may not show a persistent
* notification in that case.
*/
private suspend fun setForegroundBestEffort(filename: String, percent: Int) {
runCatching { setForeground(foregroundInfo(filename, percent)) }
}
private fun foregroundInfo(jobId: String, filename: String, percent: Int): ForegroundInfo {
private fun foregroundInfo(filename: String, percent: Int): ForegroundInfo {
val notification = NotificationCompat.Builder(applicationContext, GaloApplication.UPLOAD_CHANNEL_ID)
.setSmallIcon(android.R.drawable.stat_sys_upload)
.setContentTitle("Загрузка в galo")
@@ -188,12 +210,19 @@ class UploadWorker(
.setProgress(100, percent, percent == 0)
.build()
return ForegroundInfo(
jobId.hashCode(),
NOTIFICATION_ID,
notification,
ServiceInfo.FOREGROUND_SERVICE_TYPE_DATA_SYNC,
)
}
/** True when the active network is unmetered (Wi-Fi/ethernet). */
private fun isUnmetered(): Boolean {
val manager = applicationContext.getSystemService(ConnectivityManager::class.java) ?: return true
val capabilities = manager.getNetworkCapabilities(manager.activeNetwork) ?: return false
return capabilities.hasCapability(NetworkCapabilities.NET_CAPABILITY_NOT_METERED)
}
private fun describe(e: Exception): String = when (e) {
is ApiException -> ErrorMapper.userMessage(e.error)
else -> e.message ?: "Ошибка загрузки"
@@ -202,10 +231,8 @@ class UploadWorker(
private fun percent(sent: Long, total: Long): Int =
if (total > 0) ((sent * 100) / total).toInt().coerceIn(0, 100) else 0
@Volatile
private var lastPercent = -1
private companion object {
const val MAX_ATTEMPTS = 3
const val NOTIFICATION_ID = 42
}
}
@@ -26,6 +26,7 @@ import com.galo.android.data.sync.ApiOriginalFetcher
import com.galo.android.data.sync.ApiSyncRemote
import com.galo.android.data.sync.DeviceLocalAssetProvider
import com.galo.android.data.sync.DeviceMediaObserver
import com.galo.android.data.sync.MirrorManager
import com.galo.android.data.sync.OffloadManager
import com.galo.android.data.sync.RepositorySyncUploadQueue
import com.galo.android.data.sync.RestoreCoordinator
@@ -95,10 +96,12 @@ class AppContainer(context: Context) {
val syncRepository = SyncRepository(database.syncDao(), deviceSource)
val syncRemote: SyncRemote = ApiSyncRemote(api)
val syncRemote: SyncRemote = ApiSyncRemote(api, apiClient)
val offloadManager = OffloadManager(appContext, database.syncDao())
val mirrorManager = MirrorManager(appContext, syncRemote, deviceSource)
val syncCoordinator = SyncCoordinator(
dao = database.syncDao(),
local = DeviceLocalAssetProvider(appContext, database.syncDao(), deviceSource),
@@ -108,6 +111,7 @@ class AppContainer(context: Context) {
wifiOnly = { settingsState?.wifiOnlyUploads == true },
),
offloader = offloadManager,
mirror = mirrorManager,
)
val restoreManager = RestoreManager(
@@ -124,6 +128,16 @@ class AppContainer(context: Context) {
/** Unified per-gallery check and fix (startup + album open). */
val syncCheck = SyncCheck(syncRepository, syncRecovery, restoreCoordinator, syncCoordinator)
/**
* Watches MediaStore for this process so screens can refresh when the sync
* engine materializes or removes files. The caller stops it (return value).
*/
fun observeMediaChanges(onChanged: () -> Unit): () -> Unit {
val observer = DeviceMediaObserver(appContext, onChanged)
observer.start()
return { observer.stop() }
}
fun start() {
scope.launch {
settingsRepository.flow.collect { settings ->
@@ -140,7 +154,10 @@ class AppContainer(context: Context) {
}
}
runCatching {
DeviceMediaObserver(appContext) { runCatching { SyncScheduler.runNow(appContext) } }.start()
DeviceMediaObserver(
context = appContext,
onChanged = { runCatching { SyncScheduler.runNow(appContext) } },
).start()
}
}
@@ -22,6 +22,25 @@ enum class SyncMode {
}
}
/**
* How an element of a device album came to be: uploaded by this client, or
* materialized from a server photo during two-way synchronization. Deletion,
* offload and recovery rely on it instead of the mere presence of a server link.
*/
enum class AssetOrigin {
/** Uploaded by this client. */
OWN,
/** Downloaded from the server to mirror the target album. */
MIRROR,
;
companion object {
fun from(raw: String?): AssetOrigin =
entries.firstOrNull { it.name == raw } ?: OWN
}
}
/** Where a device asset currently lives relative to the server. */
enum class LocalAssetState {
/** Original present locally, nothing uploaded yet. */
@@ -71,4 +90,6 @@ data class DeviceAsset(
val serverPhotoId: String?,
val uploadedAt: Long?,
val state: LocalAssetState,
/** Uploaded by this client or materialized from the server. */
val origin: AssetOrigin = AssetOrigin.OWN,
)
@@ -120,6 +120,9 @@ const val TAG_FREE_BANNER = "device_free_banner"
const val TAG_FREE_BANNER_BUTTON = "device_free_banner_button"
const val TAG_NEW_ALBUM_NAME = "device_new_album_name"
const val TAG_CREATE_TARGET = "device_create_target"
const val TAG_MIRROR_SCOPE = "device_mirror_scope"
const val TAG_MIRROR_SCOPE_ONLY_NEW = "device_mirror_scope_only_new"
const val TAG_MIRROR_SCOPE_ALL = "device_mirror_scope_all"
@OptIn(ExperimentalMaterial3Api::class)
@Composable
@@ -136,6 +139,7 @@ fun DeviceAlbumScreen(
val state by viewModel.state.collectAsStateWithLifecycle()
val role by viewModel.role.collectAsStateWithLifecycle()
val notice by viewModel.notice.collectAsStateWithLifecycle()
val mirrorChoice by viewModel.mirrorChoice.collectAsStateWithLifecycle()
val prefs by viewModel.preferences.collectAsStateWithLifecycle()
val meta by viewModel.albumMeta.collectAsStateWithLifecycle()
var viewerIndex by remember { mutableStateOf<Int?>(null) }
@@ -281,6 +285,14 @@ fun DeviceAlbumScreen(
onDismiss = { deletePhotoId = null },
)
}
if (mirrorChoice) {
MirrorScopeDialog(
onOnlyNew = { viewModel.chooseMirrorScope(downloadAll = false) },
onDownloadAll = { viewModel.chooseMirrorScope(downloadAll = true) },
onDismiss = viewModel::dismissMirrorChoice,
)
}
}
@OptIn(ExperimentalMaterial3Api::class)
@@ -835,10 +847,44 @@ private fun LostCopyBanner(count: Int) {
}
}
/**
* First-pass choice for two-way synchronization: only photos that appear from now
* on, or the whole current server album.
*/
@Composable
fun MirrorScopeDialog(
onOnlyNew: () -> Unit,
onDownloadAll: () -> Unit,
onDismiss: () -> Unit,
) {
AlertDialog(
onDismissRequest = onDismiss,
modifier = Modifier.testTag(TAG_MIRROR_SCOPE),
title = { Text("Скачать содержимое альбома?") },
text = {
Text(
"Новые фото с сервера будут появляться здесь автоматически. " +
"Скачать и уже существующие фото целевого альбома?",
)
},
confirmButton = {
TextButton(
onClick = onDownloadAll,
modifier = Modifier.testTag(TAG_MIRROR_SCOPE_ALL),
) { Text("Скачать всё") }
},
dismissButton = {
TextButton(
onClick = onOnlyNew,
modifier = Modifier.testTag(TAG_MIRROR_SCOPE_ONLY_NEW),
) { Text("Только новые") }
},
)
}
/** Confirms a mode change and explains what the mode does. */
@Composable
fun ModeConfirmDialog(
mode: SyncMode,
fun ModeConfirmDialog( mode: SyncMode,
onConfirm: () -> Unit,
onDismiss: () -> Unit,
) {
@@ -72,6 +72,32 @@ class DeviceAlbumViewModel(
/** A one-off message for the user (for example a refused target change). */
val notice: StateFlow<String?> = _notice.asStateFlow()
private val _mirrorChoice = MutableStateFlow(false)
/** True while the first-pass choice ("only new" vs "download all") is shown. */
val mirrorChoice: StateFlow<Boolean> = _mirrorChoice.asStateFlow()
fun dismissMirrorChoice() {
_mirrorChoice.value = false
}
/**
* Sets the two-way mirroring baseline: only new server photos from now on, or
* the whole current album.
*/
fun chooseMirrorScope(downloadAll: Boolean) {
_mirrorChoice.value = false
viewModelScope.launch {
container.syncRepository.setMirrorFrom(
deviceAlbumId,
if (downloadAll) 0L else System.currentTimeMillis(),
)
reconcileNow()
}
}
private var stopMediaObserver: (() -> Unit)? = null
fun dismissNotice() {
_notice.value = null
}
@@ -112,6 +138,17 @@ class DeviceAlbumViewModel(
init {
load()
// Refresh the grid when MediaStore changes, so materialized server photos
// and removals appear without reopening the album.
stopMediaObserver = container.observeMediaChanges {
viewModelScope.launch { load() }
}
}
override fun onCleared() {
stopMediaObserver?.invoke()
stopMediaObserver = null
super.onCleared()
}
fun load() {
@@ -218,8 +255,12 @@ class DeviceAlbumViewModel(
fun deleteServerPhoto(photoId: String) {
viewModelScope.launch {
// Explicit in-app deletion: record the intent so the local mirror is not
// recreated, then remove the photo on the server.
container.syncRepository.markExplicitDelete(deviceAlbumId, photoId)
runCatching { container.api.deletePhoto(photoId) }
.onSuccess { container.photoRepository.remove(photoId) }
reconcileNow()
}
}
@@ -294,6 +335,8 @@ class DeviceAlbumViewModel(
viewModelScope.launch {
// Persist the chosen mode even before a server album is picked, so the
// choice survives leaving and reopening the album.
val hadTarget = container.syncRepository.currentLink(deviceAlbumId)
?.serverAlbumId?.isNotBlank() == true
val target = state.value.serverAlbumId
?: container.syncRepository.currentLink(deviceAlbumId)?.serverAlbumId
?: ""
@@ -305,7 +348,10 @@ class DeviceAlbumViewModel(
)
refreshRole()
// Nothing to sync until a server album is chosen.
if (target.isNotBlank()) reconcileNow()
if (target.isNotBlank()) {
if (!hadTarget) _mirrorChoice.value = true
reconcileNow()
}
}
}
@@ -325,6 +371,7 @@ class DeviceAlbumViewModel(
if (result.strandedCopies > 0) {
_notice.value = "Фото без серверной копии: ${result.strandedCopies}."
}
_mirrorChoice.value = mode != SyncMode.DISABLED
}
is RetargetResult.Refused -> _notice.value = result.reason
@@ -348,7 +395,10 @@ class DeviceAlbumViewModel(
mode = mode,
)
) {
is RetargetResult.Linked -> refreshRole()
is RetargetResult.Linked -> {
refreshRole()
_mirrorChoice.value = mode != SyncMode.DISABLED
}
is RetargetResult.Refused -> _notice.value = result.reason
}
}