outbox integration
This commit is contained in:
@@ -64,6 +64,9 @@ dependencies {
|
||||
implementation(libs.hilt.android)
|
||||
ksp(libs.hilt.compiler)
|
||||
implementation(libs.androidx.hilt.navigation.compose)
|
||||
implementation(libs.androidx.hilt.work)
|
||||
ksp(libs.androidx.hilt.compiler)
|
||||
implementation(libs.androidx.work.runtime.ktx)
|
||||
|
||||
implementation(libs.retrofit)
|
||||
implementation(libs.retrofit.converter.moshi)
|
||||
|
||||
@@ -1,11 +1,22 @@
|
||||
package com.vaultdrop.mobile
|
||||
|
||||
import android.app.Application
|
||||
import androidx.hilt.work.HiltWorkerFactory
|
||||
import androidx.work.Configuration
|
||||
import dagger.hilt.android.HiltAndroidApp
|
||||
import timber.log.Timber
|
||||
import javax.inject.Inject
|
||||
|
||||
@HiltAndroidApp
|
||||
class VaultDropApplication : Application() {
|
||||
class VaultDropApplication : Application(), Configuration.Provider {
|
||||
|
||||
@Inject
|
||||
lateinit var workerFactory: HiltWorkerFactory
|
||||
|
||||
override val workManagerConfiguration: Configuration
|
||||
get() = Configuration.Builder()
|
||||
.setWorkerFactory(workerFactory)
|
||||
.build()
|
||||
|
||||
override fun onCreate() {
|
||||
super.onCreate()
|
||||
|
||||
@@ -4,19 +4,25 @@ import androidx.room.Database
|
||||
import androidx.room.RoomDatabase
|
||||
import com.vaultdrop.mobile.data.local.dao.FileDao
|
||||
import com.vaultdrop.mobile.data.local.dao.FolderDao
|
||||
import com.vaultdrop.mobile.data.local.dao.PendingOperationDao
|
||||
import com.vaultdrop.mobile.data.local.dao.UserPreferenceDao
|
||||
import com.vaultdrop.mobile.data.local.entity.FileEntity
|
||||
import com.vaultdrop.mobile.data.local.entity.FolderEntity
|
||||
import com.vaultdrop.mobile.data.local.entity.PendingOperationEntity
|
||||
import com.vaultdrop.mobile.data.local.entity.UserPreferenceEntity
|
||||
|
||||
/*
|
||||
* DB SQLite locale, `dot.db` (même nom que la version Expo).
|
||||
* v1: folders ; v2: user_preferences ; v3: files ; v4: category sur files ;
|
||||
* v5: created_in_app sur folders ; v6: processed sur files (mode review).
|
||||
* v5: created_in_app sur folders ; v6: processed sur files (mode review) ;
|
||||
* v7: pending_operations (outbox).
|
||||
*/
|
||||
@Database(
|
||||
entities = [FolderEntity::class, UserPreferenceEntity::class, FileEntity::class],
|
||||
version = 6,
|
||||
entities = [
|
||||
FolderEntity::class, UserPreferenceEntity::class, FileEntity::class,
|
||||
PendingOperationEntity::class,
|
||||
],
|
||||
version = 7,
|
||||
exportSchema = false,
|
||||
)
|
||||
abstract class AppDatabase : RoomDatabase() {
|
||||
@@ -24,4 +30,5 @@ abstract class AppDatabase : RoomDatabase() {
|
||||
abstract fun folderDao(): FolderDao
|
||||
abstract fun userPreferenceDao(): UserPreferenceDao
|
||||
abstract fun fileDao(): FileDao
|
||||
abstract fun pendingOperationDao(): PendingOperationDao
|
||||
}
|
||||
+38
@@ -0,0 +1,38 @@
|
||||
package com.vaultdrop.mobile.data.local.dao
|
||||
|
||||
import androidx.room.Dao
|
||||
import androidx.room.Insert
|
||||
import androidx.room.Query
|
||||
import com.vaultdrop.mobile.data.local.entity.PendingOperationEntity
|
||||
|
||||
@Dao
|
||||
interface PendingOperationDao {
|
||||
|
||||
/** Prochaines opérations à pousser, strictement par ordre de création. */
|
||||
@Query("SELECT * FROM pending_operations WHERE status = 'pending' ORDER BY id ASC LIMIT :limit")
|
||||
suspend fun selectPending(limit: Int): List<PendingOperationEntity>
|
||||
|
||||
@Insert
|
||||
suspend fun insert(op: PendingOperationEntity): Long
|
||||
|
||||
@Query("UPDATE pending_operations SET status = 'synced', updated_at = :now WHERE id = :id")
|
||||
suspend fun markSynced(id: Long, now: Long)
|
||||
|
||||
/** Échec avec bump du compteur de tentatives (dead-letter après MAX_PENDING_ATTEMPTS). */
|
||||
@Query("UPDATE pending_operations SET status = 'failed', attempts = attempts + 1, updated_at = :now WHERE id = :id")
|
||||
suspend fun markFailed(id: Long, now: Long)
|
||||
|
||||
/**
|
||||
* Invalidation en cascade : un `create_resource` définitivement échoué rend
|
||||
* fautives toutes les ops en attente qui ciblent la même ressource (filles).
|
||||
*/
|
||||
@Query("UPDATE pending_operations SET status = 'failed', updated_at = :now WHERE resource_id = :resourceId AND status = 'pending'")
|
||||
suspend fun failDescendants(resourceId: String, now: Long)
|
||||
|
||||
/** Rétention : purge des ops synchronisées plus anciennes que `cutoff`. */
|
||||
@Query("DELETE FROM pending_operations WHERE status = 'synced' AND updated_at < :cutoff")
|
||||
suspend fun purgeSynced(cutoff: Long)
|
||||
|
||||
@Query("SELECT COUNT(*) FROM pending_operations WHERE status = 'pending'")
|
||||
suspend fun countPending(): Int
|
||||
}
|
||||
+73
@@ -0,0 +1,73 @@
|
||||
package com.vaultdrop.mobile.data.local.entity
|
||||
|
||||
import androidx.room.ColumnInfo
|
||||
import androidx.room.Entity
|
||||
import androidx.room.Index
|
||||
import androidx.room.PrimaryKey
|
||||
|
||||
/**
|
||||
* Outbox locale (transactional outbox pattern) — file des mutations locales
|
||||
* à pousser vers le serveur via `POST /sync/ops`.
|
||||
*
|
||||
* `id` = PK auto-incr Room (interne : tri, statut) ; `operationId` = UUID
|
||||
* 32-hex généré côté client, c'est la valeur exposée en `operation_id` du
|
||||
* protocole (cf. docs/api-v1.md §6.1). L'idempotence serveur s'appuie sur
|
||||
* `UNIQUE(device_id, operation_id)`.
|
||||
*
|
||||
* `payload` = métadonnées JSON légères uniquement — jamais de blobs binaires
|
||||
* (l'upload physique reste un pipeline multipart séparé, hors outbox).
|
||||
*/
|
||||
@Entity(
|
||||
tableName = "pending_operations",
|
||||
indices = [
|
||||
Index(value = ["operation_id"], unique = true),
|
||||
Index(value = ["status"]),
|
||||
Index(value = ["resource_id"]),
|
||||
],
|
||||
)
|
||||
data class PendingOperationEntity(
|
||||
@PrimaryKey(autoGenerate = true)
|
||||
val id: Long = 0L,
|
||||
@ColumnInfo(name = "operation_id")
|
||||
val operationId: String,
|
||||
@ColumnInfo(name = "ref_type")
|
||||
val refType: String? = null,
|
||||
@ColumnInfo(name = "ref_id")
|
||||
val refId: Long? = null,
|
||||
@ColumnInfo(name = "resource_id")
|
||||
val resourceId: String? = null,
|
||||
@ColumnInfo(name = "resource_type")
|
||||
val resourceType: String? = null,
|
||||
@ColumnInfo(name = "operation")
|
||||
val operation: String,
|
||||
@ColumnInfo(name = "payload")
|
||||
val payload: String,
|
||||
@ColumnInfo(name = "status")
|
||||
val status: String = PendingOpStatus.PENDING,
|
||||
@ColumnInfo(name = "attempts")
|
||||
val attempts: Int = 0,
|
||||
@ColumnInfo(name = "created_at")
|
||||
val createdAt: Long,
|
||||
@ColumnInfo(name = "updated_at")
|
||||
val updatedAt: Long,
|
||||
)
|
||||
|
||||
/** Statuts d'une op outbox — `pushStatus` du contrat sync (pending/synced/failed). */
|
||||
object PendingOpStatus {
|
||||
const val PENDING = "pending"
|
||||
const val SYNCED = "synced"
|
||||
const val FAILED = "failed"
|
||||
}
|
||||
|
||||
/** Types d'opérations de l'outbox — miroir de `PendingOperationType` (docs/api-v1.md §6.1). */
|
||||
object PendingOperationType {
|
||||
const val CREATE_RESOURCE = "create_resource"
|
||||
const val UPDATE_METADATA = "update_metadata"
|
||||
const val DELETE_RESOURCE = "delete_resource"
|
||||
const val MOVE_RESOURCE = "move_resource"
|
||||
const val SHARE = "share"
|
||||
const val REVOKE_SHARE = "revoke_share"
|
||||
const val UPDATE_SHARE = "update_share"
|
||||
const val CREATE_LINK = "create_link"
|
||||
const val REVOKE_LINK = "revoke_link"
|
||||
}
|
||||
+33
-1
@@ -17,6 +17,9 @@ import androidx.sqlite.db.SupportSQLiteDatabase
|
||||
* v6 : ajout colonne `processed` sur `files` (mode review « traiter »). Le
|
||||
* backlog existant est marqué traité à la migration : seuls les fichiers
|
||||
* découverts après la mise à jour entrent dans la file de review.
|
||||
* v7 : table `pending_operations` (outbox) — file des mutations locales à
|
||||
* pousser vers `POST /sync/ops` (uuid 32-hex client-generated, cf.
|
||||
* docs/api-v1.md §6.1).
|
||||
*/
|
||||
object Migrations {
|
||||
|
||||
@@ -87,5 +90,34 @@ object Migrations {
|
||||
}
|
||||
}
|
||||
|
||||
val ALL: Array<Migration> = arrayOf(MIGRATION_1_2, MIGRATION_2_3, MIGRATION_3_4, MIGRATION_4_5, MIGRATION_5_6)
|
||||
private val MIGRATION_6_7 = object : Migration(6, 7) {
|
||||
override fun migrate(db: SupportSQLiteDatabase) {
|
||||
db.execSQL(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS `pending_operations` (
|
||||
`id` INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
|
||||
`operation_id` TEXT NOT NULL,
|
||||
`ref_type` TEXT,
|
||||
`ref_id` INTEGER,
|
||||
`resource_id` TEXT,
|
||||
`resource_type` TEXT,
|
||||
`operation` TEXT NOT NULL,
|
||||
`payload` TEXT NOT NULL,
|
||||
`status` TEXT NOT NULL DEFAULT 'pending'
|
||||
CHECK (status IN ('pending', 'synced', 'failed')),
|
||||
`attempts` INTEGER NOT NULL DEFAULT 0,
|
||||
`created_at` INTEGER NOT NULL,
|
||||
`updated_at` INTEGER NOT NULL
|
||||
)
|
||||
""".trimIndent(),
|
||||
)
|
||||
db.execSQL("CREATE UNIQUE INDEX IF NOT EXISTS `index_pending_operations_operation_id` ON `pending_operations` (`operation_id`)")
|
||||
db.execSQL("CREATE INDEX IF NOT EXISTS `index_pending_operations_status` ON `pending_operations` (`status`)")
|
||||
db.execSQL("CREATE INDEX IF NOT EXISTS `index_pending_operations_resource_id` ON `pending_operations` (`resource_id`)")
|
||||
}
|
||||
}
|
||||
|
||||
val ALL: Array<Migration> = arrayOf(
|
||||
MIGRATION_1_2, MIGRATION_2_3, MIGRATION_3_4, MIGRATION_4_5, MIGRATION_5_6, MIGRATION_6_7,
|
||||
)
|
||||
}
|
||||
@@ -9,6 +9,10 @@ import com.vaultdrop.mobile.data.remote.dto.FileDto
|
||||
import com.vaultdrop.mobile.data.remote.dto.FolderDto
|
||||
import com.vaultdrop.mobile.data.remote.dto.LoginRequestDto
|
||||
import com.vaultdrop.mobile.data.remote.dto.LoginResponseDto
|
||||
import com.vaultdrop.mobile.data.remote.dto.ResourcePermissionDto
|
||||
import com.vaultdrop.mobile.data.remote.dto.SyncOpDto
|
||||
import com.vaultdrop.mobile.data.remote.dto.SyncOpsRequest
|
||||
import com.vaultdrop.mobile.data.remote.dto.SyncOpsResult
|
||||
import okio.IOException
|
||||
import retrofit2.Response
|
||||
import timber.log.Timber
|
||||
@@ -59,6 +63,14 @@ class ApiClient @Inject constructor(
|
||||
unwrap({ apiService.login(LoginRequestDto(username, password, deviceId)) },
|
||||
skipUnauthorizedHandling = true)
|
||||
|
||||
/** Push outbox : applique un batch d'opérations, séquentiel et idempotent. */
|
||||
suspend fun syncOps(operations: List<SyncOpDto>): SyncOpsResult =
|
||||
unwrap({ apiService.syncOps(SyncOpsRequest(operations)) })
|
||||
|
||||
/** Snapshot des permissions effectives (delta si `after` ms fourni). */
|
||||
suspend fun syncPermissions(after: Long? = null): List<ResourcePermissionDto> =
|
||||
unwrap({ apiService.syncPermissions(after) })
|
||||
|
||||
private suspend fun <T> unwrap(
|
||||
call: suspend () -> Response<ApiEnvelope<T>>,
|
||||
skipUnauthorizedHandling: Boolean = false,
|
||||
|
||||
@@ -6,6 +6,9 @@ import com.vaultdrop.mobile.data.remote.dto.FileDto
|
||||
import com.vaultdrop.mobile.data.remote.dto.FolderDto
|
||||
import com.vaultdrop.mobile.data.remote.dto.LoginRequestDto
|
||||
import com.vaultdrop.mobile.data.remote.dto.LoginResponseDto
|
||||
import com.vaultdrop.mobile.data.remote.dto.ResourcePermissionDto
|
||||
import com.vaultdrop.mobile.data.remote.dto.SyncOpsRequest
|
||||
import com.vaultdrop.mobile.data.remote.dto.SyncOpsResult
|
||||
import retrofit2.Response
|
||||
import retrofit2.http.Body
|
||||
import retrofit2.http.GET
|
||||
@@ -41,4 +44,16 @@ interface ApiService {
|
||||
suspend fun login(
|
||||
@Body body: LoginRequestDto,
|
||||
): Response<ApiEnvelope<LoginResponseDto>>
|
||||
|
||||
/** Outbox client→serveur : applique un batch séquentiel, idempotent par device. */
|
||||
@POST("sync/ops")
|
||||
suspend fun syncOps(
|
||||
@Body body: SyncOpsRequest,
|
||||
): Response<ApiEnvelope<SyncOpsResult>>
|
||||
|
||||
/** Snapshot des permissions effectives (delta si `after` fourni, ms epoch). */
|
||||
@GET("sync/permissions")
|
||||
suspend fun syncPermissions(
|
||||
@Query("after") after: Long? = null,
|
||||
): Response<ApiEnvelope<List<ResourcePermissionDto>>>
|
||||
}
|
||||
@@ -80,4 +80,45 @@ data class LoginResponseDto(
|
||||
@Json(name = "token") val token: String,
|
||||
@Json(name = "expires_at") val expiresAt: Long,
|
||||
@Json(name = "user") val user: UserDto,
|
||||
)
|
||||
|
||||
/** Outbox — une op du batch `POST /sync/ops` (docs/api-v1.md §6.1). */
|
||||
data class SyncOpDto(
|
||||
@Json(name = "operation_id") val operationId: String,
|
||||
@Json(name = "ref_type") val refType: String? = null,
|
||||
@Json(name = "ref_id") val refId: Long? = null,
|
||||
@Json(name = "resource_id") val resourceId: String? = null,
|
||||
@Json(name = "resource_type") val resourceType: String? = null,
|
||||
@Json(name = "operation") val operation: String,
|
||||
@Json(name = "payload") val payload: Map<String, Any?>? = null,
|
||||
)
|
||||
|
||||
data class SyncOpsRequest(
|
||||
@Json(name = "operations") val operations: List<SyncOpDto>,
|
||||
)
|
||||
|
||||
/** Réponse de `POST /sync/ops` : `applied` = index de la prochaine op à envoyer. */
|
||||
data class SyncOpsResult(
|
||||
@Json(name = "applied") val applied: Int,
|
||||
@Json(name = "failed") val failed: SyncFailedDto? = null,
|
||||
)
|
||||
|
||||
/** Première erreur non-idempotente du batch (arrêt du serveur). */
|
||||
data class SyncFailedDto(
|
||||
@Json(name = "operation_id") val operationId: String,
|
||||
@Json(name = "code") val code: String,
|
||||
@Json(name = "message") val message: String,
|
||||
)
|
||||
|
||||
/** Snapshot `GET /sync/permissions` (docs/api-v1.md §6.2). */
|
||||
data class ResourcePermissionDto(
|
||||
@Json(name = "resource_id") val resourceId: String,
|
||||
@Json(name = "resourceType") val resourceType: String,
|
||||
@Json(name = "effectiveAccess") val effectiveAccess: String,
|
||||
@Json(name = "inherit") val inherit: Boolean,
|
||||
@Json(name = "ownerId") val ownerId: String? = null,
|
||||
@Json(name = "sharedById") val sharedById: Any? = null,
|
||||
@Json(name = "expiresAt") val expiresAt: Any? = null,
|
||||
@Json(name = "cachedAt") val cachedAt: Long,
|
||||
@Json(name = "updatedAt") val updatedAt: Long,
|
||||
)
|
||||
+31
-3
@@ -1,5 +1,7 @@
|
||||
package com.vaultdrop.mobile.data.repository
|
||||
|
||||
import androidx.room.withTransaction
|
||||
import com.vaultdrop.mobile.data.local.AppDatabase
|
||||
import com.vaultdrop.mobile.data.local.dao.FileDao
|
||||
import com.vaultdrop.mobile.data.local.entity.FileEntity
|
||||
import com.vaultdrop.mobile.data.local.entity.FileStatus
|
||||
@@ -17,12 +19,17 @@ import javax.inject.Singleton
|
||||
*
|
||||
* Les fichiers renvoyés par le serveur sont cloud-only (uri = NULL) :
|
||||
* aucune copie physique locale, mirror de `saveFile(..., syncStatus='cloud')`.
|
||||
*
|
||||
* L'antichambre outbox (`create_resource`/`move_resource`) est écrite dans la
|
||||
* MÊME transaction que la mutation Room (pattern transactional outbox).
|
||||
*/
|
||||
@Singleton
|
||||
class FileRepository @Inject constructor(
|
||||
private val fileDao: FileDao,
|
||||
private val apiClient: ApiClient,
|
||||
private val generateId: GenerateId,
|
||||
private val appDatabase: AppDatabase,
|
||||
private val outboxRepository: OutboxRepository,
|
||||
) {
|
||||
|
||||
/** Fichiers visibles du dossier, locaux + cloud — source de l'UI. */
|
||||
@@ -97,7 +104,20 @@ class FileRepository @Inject constructor(
|
||||
addedAt = existing?.addedAt ?: now,
|
||||
updatedAt = now,
|
||||
)
|
||||
fileDao.upsert(entity)
|
||||
appDatabase.withTransaction {
|
||||
fileDao.upsert(entity)
|
||||
if (existing == null) {
|
||||
// Nouveau fichier physique → le pousser vers le serveur (métadonnées).
|
||||
outboxRepository.enqueueCreateResource(
|
||||
resourceId = entity.resourceId,
|
||||
resourceType = "file",
|
||||
name = entity.name,
|
||||
parentResourceId = entity.folderResourceId,
|
||||
mimeType = entity.mimeType,
|
||||
extension = entity.extension,
|
||||
)
|
||||
}
|
||||
}
|
||||
return entity
|
||||
}
|
||||
|
||||
@@ -108,8 +128,16 @@ class FileRepository @Inject constructor(
|
||||
/** Met à jour la cible dossier d'un fichier (déplacement local / cloud-only). */
|
||||
suspend fun applyMove(resourceId: String, folderId: String, newUri: String?) {
|
||||
val now = System.currentTimeMillis()
|
||||
fileDao.moveToFolder(listOf(resourceId), folderId, now)
|
||||
if (newUri != null) fileDao.updateUri(resourceId, newUri)
|
||||
appDatabase.withTransaction {
|
||||
fileDao.moveToFolder(listOf(resourceId), folderId, now)
|
||||
if (newUri != null) fileDao.updateUri(resourceId, newUri)
|
||||
// Déplacement reflété localement ←→ poussé vers le serveur.
|
||||
outboxRepository.enqueueMoveResource(
|
||||
resourceId = resourceId,
|
||||
resourceType = "file",
|
||||
toFolderResourceId = folderId,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
+19
-1
@@ -1,5 +1,7 @@
|
||||
package com.vaultdrop.mobile.data.repository
|
||||
|
||||
import androidx.room.withTransaction
|
||||
import com.vaultdrop.mobile.data.local.AppDatabase
|
||||
import com.vaultdrop.mobile.data.local.dao.FolderDao
|
||||
import com.vaultdrop.mobile.data.local.entity.FolderEntity
|
||||
import com.vaultdrop.mobile.data.local.entity.FolderStatus
|
||||
@@ -14,6 +16,9 @@ import javax.inject.Singleton
|
||||
/**
|
||||
* Miroir de `services/db/repositories/folders.ts`. Local-first : l'UI lit Room
|
||||
* ; le réseau sert de source de rafraîchissement (snapshot cloud).
|
||||
*
|
||||
* L'antichambre outbox (`create_resource` pour un dossier SAF nouvellement
|
||||
* découvert/créé) est écrite dans la MÊME transaction que l'upsert Room.
|
||||
*/
|
||||
@Singleton
|
||||
class FolderRepository @Inject constructor(
|
||||
@@ -21,6 +26,8 @@ class FolderRepository @Inject constructor(
|
||||
private val apiClient: ApiClient,
|
||||
private val generateId: GenerateId,
|
||||
private val deviceIdentity: DeviceIdentity,
|
||||
private val appDatabase: AppDatabase,
|
||||
private val outboxRepository: OutboxRepository,
|
||||
) {
|
||||
|
||||
fun observeRootFolders(): Flow<List<FolderEntity>> = folderDao.observeRootFolders()
|
||||
@@ -103,7 +110,18 @@ class FolderRepository @Inject constructor(
|
||||
addedAt = existing?.addedAt ?: now,
|
||||
updatedAt = now,
|
||||
)
|
||||
folderDao.upsert(entity)
|
||||
appDatabase.withTransaction {
|
||||
folderDao.upsert(entity)
|
||||
if (existing == null) {
|
||||
// Nouvelle ressource physique → la pousser vers le serveur.
|
||||
outboxRepository.enqueueCreateResource(
|
||||
resourceId = entity.resourceId,
|
||||
resourceType = "folder",
|
||||
name = entity.name,
|
||||
parentResourceId = entity.parentResourceId,
|
||||
)
|
||||
}
|
||||
}
|
||||
return entity
|
||||
}
|
||||
|
||||
|
||||
+108
@@ -0,0 +1,108 @@
|
||||
package com.vaultdrop.mobile.data.repository
|
||||
|
||||
import com.squareup.moshi.JsonAdapter
|
||||
import com.squareup.moshi.Moshi
|
||||
import com.squareup.moshi.Types
|
||||
import com.vaultdrop.mobile.data.local.dao.PendingOperationDao
|
||||
import com.vaultdrop.mobile.data.local.entity.PendingOpStatus
|
||||
import com.vaultdrop.mobile.data.local.entity.PendingOperationEntity
|
||||
import com.vaultdrop.mobile.data.local.entity.PendingOperationType
|
||||
import com.vaultdrop.mobile.domain.GenerateId
|
||||
import javax.inject.Inject
|
||||
import javax.inject.Singleton
|
||||
|
||||
/**
|
||||
* Antichambre de l'outbox : enregistre les mutations locales (métadonnées JSON
|
||||
* uniquement) à pousser vers `POST /sync/ops`.
|
||||
*
|
||||
* **Contrat d'atomicité** : `enqueue` doit être appelé dans le MÊME bloc
|
||||
* `AppDatabase.withTransaction { … }` que la mutation Room correspondante —
|
||||
* la file et l'état local évoluent ensemble (pattern transactional outbox).
|
||||
*/
|
||||
@Singleton
|
||||
class OutboxRepository @Inject constructor(
|
||||
private val pendingOperationDao: PendingOperationDao,
|
||||
private val generateId: GenerateId,
|
||||
moshi: Moshi,
|
||||
) {
|
||||
|
||||
private val payloadAdapter: JsonAdapter<Map<String, Any?>> = moshi.adapter(
|
||||
Types.newParameterizedType(Map::class.java, String::class.java, Any::class.java),
|
||||
)
|
||||
|
||||
/**
|
||||
* Enregistre une op à pousser. Retourne l'`operationId` 32-hex (exposé en
|
||||
* `operation_id` du protocole). À appeler dans un `withTransaction`.
|
||||
*/
|
||||
suspend fun enqueue(
|
||||
operation: String,
|
||||
resourceId: String,
|
||||
resourceType: String,
|
||||
payload: Map<String, Any?> = emptyMap(),
|
||||
refType: String? = null,
|
||||
refId: Long? = null,
|
||||
): String {
|
||||
val operationId = generateId.newResourceId()
|
||||
val now = System.currentTimeMillis()
|
||||
pendingOperationDao.insert(
|
||||
PendingOperationEntity(
|
||||
operationId = operationId,
|
||||
refType = refType,
|
||||
refId = refId,
|
||||
resourceId = resourceId,
|
||||
resourceType = resourceType,
|
||||
operation = operation,
|
||||
payload = payloadAdapter.toJson(payload),
|
||||
status = PendingOpStatus.PENDING,
|
||||
createdAt = now,
|
||||
updatedAt = now,
|
||||
),
|
||||
)
|
||||
return operationId
|
||||
}
|
||||
|
||||
/** Raccourci `create_resource` (folder/file) — docs/api-v1.md §6.1. */
|
||||
suspend fun enqueueCreateResource(
|
||||
resourceId: String,
|
||||
resourceType: String,
|
||||
name: String,
|
||||
parentResourceId: String? = null,
|
||||
mimeType: String? = null,
|
||||
extension: String? = null,
|
||||
): String = enqueue(
|
||||
operation = PendingOperationType.CREATE_RESOURCE,
|
||||
resourceId = resourceId,
|
||||
resourceType = resourceType,
|
||||
payload = buildMap {
|
||||
put("name", name)
|
||||
parentResourceId?.let { put("parentResourceId", it) }
|
||||
mimeType?.let { put("mimeType", it) }
|
||||
extension?.let { put("extension", it) }
|
||||
},
|
||||
)
|
||||
|
||||
/** Raccourci `move_resource` — payload `{ toFolderResourceId }`. */
|
||||
suspend fun enqueueMoveResource(
|
||||
resourceId: String,
|
||||
resourceType: String,
|
||||
toFolderResourceId: String,
|
||||
): String = enqueue(
|
||||
operation = PendingOperationType.MOVE_RESOURCE,
|
||||
resourceId = resourceId,
|
||||
resourceType = resourceType,
|
||||
payload = mapOf("toFolderResourceId" to toFolderResourceId),
|
||||
)
|
||||
|
||||
/** Raccourci `delete_resource` — idempotent côté serveur. */
|
||||
suspend fun enqueueDeleteResource(
|
||||
resourceId: String,
|
||||
resourceType: String,
|
||||
): String = enqueue(
|
||||
operation = PendingOperationType.DELETE_RESOURCE,
|
||||
resourceId = resourceId,
|
||||
resourceType = resourceType,
|
||||
)
|
||||
|
||||
/** Nombre d'ops en attente de push (stats UI optionnelles). */
|
||||
suspend fun countPending(): Int = pendingOperationDao.countPending()
|
||||
}
|
||||
@@ -5,6 +5,7 @@ import androidx.room.Room
|
||||
import com.vaultdrop.mobile.data.local.AppDatabase
|
||||
import com.vaultdrop.mobile.data.local.dao.FileDao
|
||||
import com.vaultdrop.mobile.data.local.dao.FolderDao
|
||||
import com.vaultdrop.mobile.data.local.dao.PendingOperationDao
|
||||
import com.vaultdrop.mobile.data.local.dao.UserPreferenceDao
|
||||
import com.vaultdrop.mobile.data.local.migration.Migrations
|
||||
import dagger.Module
|
||||
@@ -33,4 +34,7 @@ object DatabaseModule {
|
||||
|
||||
@Provides
|
||||
fun provideFileDao(db: AppDatabase): FileDao = db.fileDao()
|
||||
|
||||
@Provides
|
||||
fun providePendingOperationDao(db: AppDatabase): PendingOperationDao = db.pendingOperationDao()
|
||||
}
|
||||
@@ -22,7 +22,7 @@ import javax.inject.Singleton
|
||||
* bas de gamme : métadonnée seule (`folderResourceId`), la marche suivante
|
||||
* re-réconciliera.
|
||||
* - fichier cloud-only → mise à jour de la métadonnée de dossier uniquement
|
||||
* (pas de poussée serveur en V1 — pas d'outbox).
|
||||
* (`applyMove` journalise aussi un `move_resource` dans l'outbox).
|
||||
*/
|
||||
@Singleton
|
||||
class FileMover @Inject constructor(
|
||||
|
||||
+12
-1
@@ -4,8 +4,11 @@ import android.content.ContentResolver
|
||||
import android.content.Context
|
||||
import android.net.Uri
|
||||
import android.provider.DocumentsContract
|
||||
import androidx.room.withTransaction
|
||||
import com.vaultdrop.mobile.data.local.AppDatabase
|
||||
import com.vaultdrop.mobile.data.local.entity.FileEntity
|
||||
import com.vaultdrop.mobile.data.repository.FileRepository
|
||||
import com.vaultdrop.mobile.data.repository.OutboxRepository
|
||||
import dagger.hilt.android.qualifiers.ApplicationContext
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.withContext
|
||||
@@ -20,6 +23,8 @@ import javax.inject.Singleton
|
||||
* disparaît aussi de l'arborescence (et donc du prochain sync). En cas de
|
||||
* succès, la ligne Room est masquée (`exists = 0`) : cohérent avec la
|
||||
* réconciliation (jamais de DELETE SQL), idempotent face à la marche 30s.
|
||||
* La suppression est aussi journalisée dans l'outbox (`delete_resource`),
|
||||
* dans la même transaction que le masquage Room.
|
||||
*
|
||||
* Retourne `false` si le provider refuse la suppression (permission ou pas de
|
||||
* geste delete) : dans ce cas le fichier reste dans la file de review.
|
||||
@@ -28,6 +33,8 @@ import javax.inject.Singleton
|
||||
class SafFileDeleter @Inject constructor(
|
||||
@ApplicationContext private val context: Context,
|
||||
private val fileRepository: FileRepository,
|
||||
private val outboxRepository: OutboxRepository,
|
||||
private val appDatabase: AppDatabase,
|
||||
) {
|
||||
|
||||
private val resolver: ContentResolver get() = context.contentResolver
|
||||
@@ -39,8 +46,12 @@ class SafFileDeleter @Inject constructor(
|
||||
DocumentsContract.deleteDocument(resolver, Uri.parse(uri))
|
||||
}.onSuccess { deleted ->
|
||||
if (deleted) {
|
||||
val now = System.currentTimeMillis()
|
||||
appDatabase.withTransaction {
|
||||
fileRepository.markMissing(file.resourceId, now)
|
||||
outboxRepository.enqueueDeleteResource(file.resourceId, "file")
|
||||
}
|
||||
Timber.d("deleted %s", uri)
|
||||
fileRepository.markMissing(file.resourceId, System.currentTimeMillis())
|
||||
} else {
|
||||
Timber.w("deleteDocument returned false for %s", uri)
|
||||
}
|
||||
|
||||
+155
@@ -0,0 +1,155 @@
|
||||
package com.vaultdrop.mobile.features.sync
|
||||
|
||||
import android.content.Context
|
||||
import androidx.hilt.work.HiltWorker
|
||||
import androidx.work.BackoffPolicy
|
||||
import androidx.work.Constraints
|
||||
import androidx.work.CoroutineWorker
|
||||
import androidx.work.ExistingWorkPolicy
|
||||
import androidx.work.NetworkType
|
||||
import androidx.work.OneTimeWorkRequestBuilder
|
||||
import androidx.work.WorkManager
|
||||
import androidx.work.WorkerParameters
|
||||
import com.squareup.moshi.JsonAdapter
|
||||
import com.squareup.moshi.Moshi
|
||||
import com.squareup.moshi.Types
|
||||
import com.vaultdrop.mobile.auth.TokenProvider
|
||||
import com.vaultdrop.mobile.data.local.dao.PendingOperationDao
|
||||
import com.vaultdrop.mobile.data.local.entity.PendingOperationEntity
|
||||
import com.vaultdrop.mobile.data.local.entity.PendingOperationType
|
||||
import com.vaultdrop.mobile.data.remote.ApiClient
|
||||
import com.vaultdrop.mobile.data.remote.ApiException
|
||||
import com.vaultdrop.mobile.data.remote.dto.SyncOpDto
|
||||
import dagger.assisted.Assisted
|
||||
import dagger.assisted.AssistedInject
|
||||
import java.util.concurrent.TimeUnit
|
||||
import timber.log.Timber
|
||||
|
||||
/**
|
||||
* Worker de push de l'outbox : applique les batchs `pending_operations` vers
|
||||
* `POST /sync/ops`, séquentiellement et dans l'ordre d'entrée en file.
|
||||
*
|
||||
* Garanties :
|
||||
* - **single-flight** : enregistré via `enqueueUniqueWork(KEEP)` — jamais deux
|
||||
* workers parallèles (ordonnancement chronologique préservé) ;
|
||||
* - **no-op sans token** : mode local, rien à pousser (contrat syncOutbox) ;
|
||||
* - **retry transient** : `NETWORK_ERROR`/5xx/timeout → `Result.retry()`
|
||||
* (backoff exponentiel) sans bump du compteur d'attempts ;
|
||||
* - **dead-letter permanent** : erreur 4xx non-idempotente → op passée en
|
||||
* `failed` immédiatement (évite d'affamer la queue : le serveur s'arrête à
|
||||
* la 1re erreur), et cascade si `create_resource` a échoué (toutes les ops
|
||||
* filles en attente sur la même resource passent `failed`) ;
|
||||
* - `attempts` est un compteur diagnostic (incrémenté à chaque dead-letter),
|
||||
* pas un seuil : une op `failed` n'est plus resélectionnée.
|
||||
* - push **avant** pull : le pull `GET /sync/permissions` (V2) s'appliquera
|
||||
* après ce worker — jamais d'écrasement d'état optimiste.
|
||||
*/
|
||||
@HiltWorker
|
||||
class OutboxSyncWorker @AssistedInject constructor(
|
||||
@Assisted appContext: Context,
|
||||
@Assisted workerParams: WorkerParameters,
|
||||
private val apiClient: ApiClient,
|
||||
private val pendingOperationDao: PendingOperationDao,
|
||||
private val tokenProvider: TokenProvider,
|
||||
moshi: Moshi,
|
||||
) : CoroutineWorker(appContext, workerParams) {
|
||||
|
||||
private val payloadAdapter: JsonAdapter<Map<String, Any?>> = moshi.adapter(
|
||||
Types.newParameterizedType(Map::class.java, String::class.java, Any::class.java),
|
||||
)
|
||||
|
||||
override suspend fun doWork(): Result {
|
||||
// Mode local : sans compte connecté, rien à pousser.
|
||||
if (tokenProvider.current == null) return Result.success()
|
||||
|
||||
while (true) {
|
||||
val pending = pendingOperationDao.selectPending(BATCH_SIZE)
|
||||
if (pending.isEmpty()) {
|
||||
pendingOperationDao.purgeSynced(System.currentTimeMillis() - PURGE_AGE_MS)
|
||||
return Result.success()
|
||||
}
|
||||
|
||||
val ops = pending.map { it.toSyncOpDto() }
|
||||
val result = try {
|
||||
apiClient.syncOps(ops)
|
||||
} catch (e: ApiException) {
|
||||
when {
|
||||
e.code == "NETWORK_ERROR" || e.httpCode >= 500 -> {
|
||||
// Transitoire : on retentera avec backoff, sans bump d'attempts.
|
||||
return Result.retry()
|
||||
}
|
||||
e.httpCode == 401 -> {
|
||||
// Token expiré/révoqué : le re-login se fera via l'UI.
|
||||
return Result.success()
|
||||
}
|
||||
else -> {
|
||||
// Erreur permanente : dead-letter la 1re op du batch.
|
||||
val now = System.currentTimeMillis()
|
||||
pendingOperationDao.markFailed(pending.first().id, now)
|
||||
continue
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
val now = System.currentTimeMillis()
|
||||
// Ops appliquées par le serveur (indices < applied).
|
||||
for (i in 0 until result.applied) {
|
||||
pendingOperationDao.markSynced(pending[i].id, now)
|
||||
Timber.d("outbox synced %s", pending[i].operationId)
|
||||
}
|
||||
|
||||
// Première erreur non-idempotente → dead-letter + cascade éventuelle.
|
||||
result.failed?.let { fail ->
|
||||
val failedIndex = pending.indexOfFirst { it.operationId == fail.operationId }
|
||||
if (failedIndex >= 0) {
|
||||
val failedOp = pending[failedIndex]
|
||||
pendingOperationDao.markFailed(failedOp.id, now)
|
||||
Timber.w("outbox failed %s (%s: %s)", fail.operationId, fail.code, fail.message)
|
||||
if (failedOp.operation == PendingOperationType.CREATE_RESOURCE) {
|
||||
failedOp.resourceId?.let { resourceId ->
|
||||
pendingOperationDao.failDescendants(resourceId, now)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Boucle : d'autres batchs attendent → traité dans le même run.
|
||||
}
|
||||
}
|
||||
|
||||
private fun PendingOperationEntity.toSyncOpDto(): SyncOpDto {
|
||||
val payloadMap = runCatching { payloadAdapter.fromJson(payload) }.getOrNull()
|
||||
return SyncOpDto(
|
||||
operationId = operationId,
|
||||
refType = refType,
|
||||
refId = refId,
|
||||
resourceId = resourceId,
|
||||
resourceType = resourceType,
|
||||
operation = operation,
|
||||
payload = payloadMap,
|
||||
)
|
||||
}
|
||||
|
||||
companion object {
|
||||
const val NAME = "outbox_sync"
|
||||
private const val BATCH_SIZE = 20
|
||||
private const val PURGE_AGE_MS = 7 * 24 * 60 * 60 * 1000L
|
||||
|
||||
/**
|
||||
* Enregistre (idempotent) le worker avec contrainte réseau + backoff.
|
||||
* `KEEP` : si un run est déjà planifié/en cours, on ne le remplace pas —
|
||||
* single-flight garanti.
|
||||
*/
|
||||
fun enqueue(context: Context) {
|
||||
val constraints = Constraints.Builder()
|
||||
.setRequiredNetworkType(NetworkType.CONNECTED)
|
||||
.build()
|
||||
val request = OneTimeWorkRequestBuilder<OutboxSyncWorker>()
|
||||
.setConstraints(constraints)
|
||||
.setBackoffCriteria(BackoffPolicy.EXPONENTIAL, 30, TimeUnit.SECONDS)
|
||||
.build()
|
||||
WorkManager.getInstance(context)
|
||||
.enqueueUniqueWork(NAME, ExistingWorkPolicy.KEEP, request)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -2,9 +2,11 @@ package com.vaultdrop.mobile.features.sync
|
||||
|
||||
import androidx.lifecycle.ViewModel
|
||||
import androidx.lifecycle.viewModelScope
|
||||
import android.content.Context
|
||||
import com.vaultdrop.mobile.data.repository.FolderRepository
|
||||
import com.vaultdrop.mobile.data.repository.SaveFolderInput
|
||||
import dagger.hilt.android.lifecycle.HiltViewModel
|
||||
import dagger.hilt.android.qualifiers.ApplicationContext
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.delay
|
||||
@@ -31,6 +33,7 @@ import javax.inject.Inject
|
||||
class SyncViewModel @Inject constructor(
|
||||
private val deviceSync: DeviceSync,
|
||||
private val folderRepository: FolderRepository,
|
||||
@ApplicationContext private val appContext: Context,
|
||||
) : ViewModel() {
|
||||
|
||||
private var loopJob: Job? = null
|
||||
@@ -54,6 +57,9 @@ class SyncViewModel @Inject constructor(
|
||||
runCatching { deviceSync.syncAll() }
|
||||
.onSuccess { results ->
|
||||
if (results.isNotEmpty()) Timber.d("syncAll: %s", results)
|
||||
// Les nouvelles ressources découvertes sont dans l'outbox
|
||||
// → drainer vers POST /sync/ops (single-flight via KEEP).
|
||||
OutboxSyncWorker.enqueue(appContext)
|
||||
}
|
||||
.onFailure { e -> Timber.w(e, "syncAll failed, retrying later") }
|
||||
delay(INTERVAL_MS)
|
||||
|
||||
@@ -1,12 +1,15 @@
|
||||
package com.vaultdrop.mobile.ui.auth
|
||||
|
||||
import android.content.Context
|
||||
import androidx.lifecycle.ViewModel
|
||||
import androidx.lifecycle.viewModelScope
|
||||
import com.vaultdrop.mobile.auth.SessionManager
|
||||
import com.vaultdrop.mobile.data.remote.ApiClient
|
||||
import com.vaultdrop.mobile.data.remote.ApiException
|
||||
import com.vaultdrop.mobile.data.repository.AuthRepository
|
||||
import com.vaultdrop.mobile.features.sync.OutboxSyncWorker
|
||||
import dagger.hilt.android.lifecycle.HiltViewModel
|
||||
import dagger.hilt.android.qualifiers.ApplicationContext
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.flow.MutableStateFlow
|
||||
import kotlinx.coroutines.flow.StateFlow
|
||||
@@ -22,6 +25,7 @@ class AuthViewModel @Inject constructor(
|
||||
private val authRepository: AuthRepository,
|
||||
private val sessionManager: SessionManager,
|
||||
private val apiClient: ApiClient,
|
||||
@ApplicationContext private val appContext: Context,
|
||||
) : ViewModel() {
|
||||
|
||||
private val _authState = MutableStateFlow<AuthState>(AuthState.Loading)
|
||||
@@ -43,6 +47,8 @@ class AuthViewModel @Inject constructor(
|
||||
else -> AuthState.SignedOut
|
||||
}
|
||||
authRepository.registerDevice()
|
||||
// Session restaurée → drainer l'outbox laissée en attente.
|
||||
OutboxSyncWorker.enqueue(appContext)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -58,6 +64,8 @@ class AuthViewModel @Inject constructor(
|
||||
sessionManager.save(response.token, response.user)
|
||||
_loginUiState.value = LoginUiState()
|
||||
_authState.value = AuthState.SignedIn(response.user)
|
||||
// Connexion réussie → pousser les mutations locales en attente.
|
||||
OutboxSyncWorker.enqueue(appContext)
|
||||
}
|
||||
.onFailure { e ->
|
||||
val error = when (e) {
|
||||
|
||||
@@ -10,6 +10,7 @@ navigationCompose = "2.8.5"
|
||||
room = "2.6.1"
|
||||
hilt = "2.53.1"
|
||||
hiltNavigationCompose = "1.2.0"
|
||||
workManager = "2.9.1"
|
||||
retrofit = "2.11.0"
|
||||
okhttp = "4.12.0"
|
||||
moshi = "1.15.1"
|
||||
@@ -37,6 +38,9 @@ androidx-room-compiler = { group = "androidx.room", name = "room-compiler", vers
|
||||
hilt-android = { group = "com.google.dagger", name = "hilt-android", version.ref = "hilt" }
|
||||
hilt-compiler = { group = "com.google.dagger", name = "hilt-android-compiler", version.ref = "hilt" }
|
||||
androidx-hilt-navigation-compose = { group = "androidx.hilt", name = "hilt-navigation-compose", version.ref = "hiltNavigationCompose" }
|
||||
androidx-hilt-work = { group = "androidx.hilt", name = "hilt-work", version.ref = "hiltNavigationCompose" }
|
||||
androidx-hilt-compiler = { group = "androidx.hilt", name = "hilt-compiler", version.ref = "hiltNavigationCompose" }
|
||||
androidx-work-runtime-ktx = { group = "androidx.work", name = "work-runtime-ktx", version.ref = "workManager" }
|
||||
retrofit = { group = "com.squareup.retrofit2", name = "retrofit", version.ref = "retrofit" }
|
||||
retrofit-converter-moshi = { group = "com.squareup.retrofit2", name = "converter-moshi", version.ref = "retrofit" }
|
||||
okhttp = { group = "com.squareup.okhttp3", name = "okhttp", version.ref = "okhttp" }
|
||||
|
||||
Reference in New Issue
Block a user