android: watch folders so new ROMs are cached automatically

app/src/main/java/com/raofflineproxy/data/CacheDao.kt

@@ -49,6 +49,8 @@ interface CacheDao {
           AND cacheKey NOT LIKE 'startsession:%'
           AND cacheKey NOT LIKE 'gameid:%'
           AND cacheKey NOT LIKE 'cachequeue:%'
+          AND cacheKey NOT LIKE 'watchfolder:%'
+          AND cacheKey NOT LIKE 'watchseen:%'
         """
     )
     suspend fun evictOlderThan(before: Long)

app/src/main/java/com/raofflineproxy/data/CacheKeys.kt

@@ -11,6 +11,8 @@ object CacheKeys {
     const val PREFIX_GAMEID = "gameid:"
     const val PREFIX_LAST_PLAYED = "lastplayed:"
     const val PREFIX_CACHE_QUEUE = "cachequeue:"
+    const val PREFIX_WATCH_FOLDER = "watchfolder:"
+    const val PREFIX_WATCH_SEEN = "watchseen:"
     const val CACHE_BUDGET = "cachebudget"
 
     fun login(user: String) = "$PREFIX_LOGIN${user.lowercase()}"
@@ -26,6 +28,9 @@ object CacheKeys {
     fun startSession(gameId: String, user: String) = "$PREFIX_STARTSESSION$gameId:${user.lowercase()}:0"
     fun lastPlayed(gameId: Int) = "$PREFIX_LAST_PLAYED$gameId"
     fun cacheQueue(hash: String) = "$PREFIX_CACHE_QUEUE${hash.lowercase()}"
+    fun watchFolder(treeUri: String) = "$PREFIX_WATCH_FOLDER$treeUri"
+    fun watchSeenPrefix(treeUri: String) = "$PREFIX_WATCH_SEEN$treeUri|"
+    fun watchSeen(treeUri: String, romPath: String) = "${watchSeenPrefix(treeUri)}$romPath"
 
     fun parseGameIdFromPatchKey(cacheKey: String): Int? =
         cacheKey.removePrefix(PREFIX_PATCH).split(":").firstOrNull()?.toIntOrNull()

app/src/main/java/com/raofflineproxy/proxy/RomScanner.kt

@@ -40,7 +40,7 @@ private const val HTTP_RETRY_AFTER_HEADER = "Retry-After"
 private const val HTTP_GET_MAX_429_RETRIES = 4
 private const val HTTP_GET_INITIAL_429_BACKOFF_MS = 2_000L
 private const val HTTP_GET_MAX_429_BACKOFF_MS = 15_000L
-private const val MAX_SCAN_ENTRIES = 5000
+internal const val MAX_SCAN_ENTRIES = 5000
 private const val MAX_SCAN_DEPTH = 12
 // RetroAchievements adds hashes over time, so a "no match" is only cached long enough to
 // stop a repeated scan of the same folder from re-querying every unsupported ROM.
@@ -699,7 +699,7 @@ internal fun String.normalizeCachedRomPath(): String =
     replace('\\', '/')
         .trim()
 
-private fun collectScannableFiles(root: DocumentFile, maxEntries: Int): List<DocumentFile> {
+internal fun collectScannableFiles(root: DocumentFile, maxEntries: Int): List<DocumentFile> {
     val result = mutableListOf<DocumentFile>()
     if (maxEntries <= 0) return result
     val stack = ArrayDeque<Pair<DocumentFile, Int>>()

app/src/main/java/com/raofflineproxy/proxy/WatchFolders.kt

@@ -0,0 +1,189 @@
+package com.raofflineproxy.proxy
+
+import android.content.Context
+import android.net.Uri
+import android.util.Log
+import androidx.core.net.toUri
+import androidx.documentfile.provider.DocumentFile
+import com.raofflineproxy.data.AppDatabase
+import com.raofflineproxy.data.CacheEntry
+import com.raofflineproxy.data.CacheKeys
+import com.raofflineproxy.data.forEachEntryByPrefix
+import com.raofflineproxy.proxy.hash.hashRomCandidates
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.map
+import org.json.JSONObject
+
+private const val TAG = "RAProxy/WatchFolders"
+
+internal const val WATCH_RESCAN_INTERVAL_MS = CACHE_BUDGET_WINDOW_MS
+
+data class WatchedFolder(
+    val uri: String,
+    val name: String,
+    val path: String?,
+    val watching: Boolean,
+    val addedAt: Long,
+    val lastScanAt: Long
+) {
+    // A clock that jumped backwards makes the folder due again instead of postponing it.
+    fun isDue(now: Long): Boolean =
+        watching && (now < lastScanAt || now - lastScanAt >= WATCH_RESCAN_INTERVAL_MS)
+
+    fun toJson(): String = JSONObject()
+        .put("uri", uri)
+        .put("name", name)
+        .put("path", path ?: JSONObject.NULL)
+        .put("watching", watching)
+        .put("addedAt", addedAt)
+        .put("lastScanAt", lastScanAt)
+        .toString()
+
+    companion object {
+        fun fromJson(body: String): WatchedFolder? = runCatching {
+            val json = JSONObject(body)
+            WatchedFolder(
+                uri = json.getString("uri").takeIf { it.isNotBlank() } ?: return null,
+                name = json.optString("name"),
+                path = json.optString("path").takeIf { !json.isNull("path") && it.isNotBlank() },
+                watching = json.optBoolean("watching", false),
+                addedAt = json.optLong("addedAt"),
+                lastScanAt = json.optLong("lastScanAt")
+            )
+        }.getOrNull()
+    }
+}
+
+internal data class FolderScan(val files: Int, val new: Int, val handled: Int, val queued: Int) {
+    val complete: Boolean get() = handled == new
+}
+
+/** Folders the user has added, stored in the cache table so that no schema change is needed.
+ *  A watched folder is rescanned once per budget window; [scanWatchedFolder] hashes and queues
+ *  only the files it has not handled before. */
+internal object WatchFolders {
+    fun observe(db: AppDatabase): Flow<List<WatchedFolder>> =
+        db.cacheDao().observeByPrefix(CacheKeys.PREFIX_WATCH_FOLDER).map { entries ->
+            entries.mapNotNull { WatchedFolder.fromJson(it.responseBody) }
+                .sortedBy { it.name.lowercase() }
+        }
+
+    suspend fun all(db: AppDatabase): List<WatchedFolder> = buildList {
+        db.cacheDao().forEachEntryByPrefix(CacheKeys.PREFIX_WATCH_FOLDER) { entry ->
+            WatchedFolder.fromJson(entry.responseBody)?.let(::add)
+        }
+    }
+
+    suspend fun due(db: AppDatabase, now: Long): List<WatchedFolder> =
+        all(db).filter { it.isDue(now) }
+
+    /** Returns false when the folder was already known; its watching state stays as it was. */
+    suspend fun add(context: Context, db: AppDatabase, treeUri: Uri, now: Long = System.currentTimeMillis()): Boolean {
+        val root = DocumentFile.fromTreeUri(context, treeUri)
+        val folder = WatchedFolder(
+            uri = treeUri.toString(),
+            name = root?.name ?: treeUri.lastPathSegment?.substringAfterLast(':').orEmpty(),
+            path = root?.let(::resolveDocumentRelativePath),
+            watching = false,
+            addedAt = now,
+            lastScanAt = 0L
+        )
+        return db.cacheDao().insertIgnore(
+            CacheEntry(cacheKey = CacheKeys.watchFolder(folder.uri), responseBody = folder.toJson())
+        ) != -1L
+    }
+
+    suspend fun setWatching(db: AppDatabase, uri: String, watching: Boolean) {
+        update(db, uri) { it.copy(watching = watching, lastScanAt = 0L) }
+    }
+
+    suspend fun remove(db: AppDatabase, uri: String) {
+        db.cacheDao().deleteByKey(CacheKeys.watchFolder(uri))
+        forgetSeen(db, seenKeys(db, uri))
+    }
+
+    suspend fun recordScan(db: AppDatabase, uri: String, scannedAt: Long) {
+        update(db, uri) { it.copy(lastScanAt = scannedAt) }
+    }
+
+    private suspend fun update(db: AppDatabase, uri: String, change: (WatchedFolder) -> WatchedFolder) {
+        val key = CacheKeys.watchFolder(uri)
+        val current = db.cacheDao().get(key)?.let { WatchedFolder.fromJson(it.responseBody) } ?: return
+        db.cacheDao().updateBody(key, change(current).toJson(), null, System.currentTimeMillis())
+    }
+
+    // The prefix lookup is a LIKE match, so "_" in a tree uri would also match neighbouring
+    // folders; the startsWith check keeps it to this one.
+    private suspend fun seenKeys(db: AppDatabase, uri: String): Set<String> {
+        val prefix = CacheKeys.watchSeenPrefix(uri)
+        return db.cacheDao().getKeysByPrefix(prefix).filterTo(mutableSetOf()) { it.startsWith(prefix) }
+    }
+
+    private suspend fun seenPaths(db: AppDatabase, uri: String): Set<String> {
+        val prefixLength = CacheKeys.watchSeenPrefix(uri).length
+        return seenKeys(db, uri).mapTo(mutableSetOf()) { it.substring(prefixLength) }
+    }
+
+    private suspend fun markSeen(db: AppDatabase, uri: String, path: String) {
+        db.cacheDao().insertIgnore(CacheEntry(cacheKey = CacheKeys.watchSeen(uri, path), responseBody = "{}"))
+    }
+
+    private suspend fun forgetSeen(db: AppDatabase, keys: Collection<String>) {
+        keys.forEach { db.cacheDao().deleteByKey(it) }
+    }
+
+    /** Hashes and queues only the files this folder has not handled before. A file counts as
+     *  handled once caching has an answer for it, including "no RetroAchievements match", so
+     *  unknown files are not hashed again on every pass. Returns null when the folder cannot
+     *  be read, e.g. because its access grant is gone. */
+    suspend fun scan(
+        context: Context,
+        db: AppDatabase,
+        folder: WatchedFolder,
+        shouldAbort: () -> Boolean,
+        now: () -> Long = System::currentTimeMillis
+    ): FolderScan? {
+        val treeUri = folder.uri.toUri()
+        val granted = context.contentResolver.persistedUriPermissions.any { it.uri == treeUri && it.isReadPermission }
+        val root = DocumentFile.fromTreeUri(context, treeUri)?.takeIf { granted && it.canRead() } ?: return null
+
+        val files = collectScannableFiles(root, MAX_SCAN_ENTRIES)
+        val romPaths = files.map(::resolveDocumentAbsolutePath)
+        val seenKeys = files.indices.map { romPaths[it] ?: files[it].uri.toString() }
+        val seen = seenPaths(db, folder.uri)
+        val knownPaths = loadCachedRomPaths(db) + CacheQueue.queuedRomPaths(db)
+        val cachedGameIds = loadCachedGameIds(db)
+
+        val unseen = files.indices.filter { seenKeys[it] !in seen }
+        val (known, fresh) = unseen.partition { romPaths[it]?.normalizeCachedRomPath() in knownPaths }
+        known.forEach { markSeen(db, folder.uri, seenKeys[it]) }
+
+        var handled = 0
+        var queued = 0
+        for (index in fresh) {
+            if (shouldAbort()) break
+            val file = files[index]
+            val candidates = runCatching { hashRomCandidates(context, file) }
+                .onFailure { Log.w(TAG, "Could not hash ${file.name}", it) }
+                .getOrDefault(emptyList())
+            if (candidates.isNotEmpty() &&
+                enqueueRomIfNeeded(db, candidates, cachedGameIds, romPaths[index], file.name ?: "")
+            ) {
+                queued++
+            }
+            markSeen(db, folder.uri, seenKeys[index])
+            handled++
+        }
+
+        // A capped listing leaves out files that still exist, so their markers must survive.
+        if (files.size < MAX_SCAN_ENTRIES) {
+            val prefix = CacheKeys.watchSeenPrefix(folder.uri)
+            val gone = (seen - seenKeys.toSet()).map { "$prefix$it" }
+            forgetSeen(db, gone)
+        }
+
+        val scan = FolderScan(files.size, fresh.size, handled, queued)
+        if (scan.complete) recordScan(db, folder.uri, now())
+        return scan
+    }
+}

app/src/main/java/com/raofflineproxy/service/CacheQueueAlarms.kt

@@ -59,6 +59,9 @@ internal val CacheQueueWakeLock = CachingWakeLock("RAOfflineProxy:cacheQueue")
  *  apart from the worker's lock, whose release at the end of every round would otherwise end it. */
 internal val BulkRunWakeLock = CachingWakeLock("RAOfflineProxy:bulkRun")
 
+/** Hashing a watched folder in the service; kept apart from the worker's lock for the same reason. */
+internal val FolderWatchWakeLock = CachingWakeLock("RAOfflineProxy:folderWatch")
+
 class CacheQueueAlarmReceiver : BroadcastReceiver() {
     // The system only holds a wake lock for onReceive, so take ours here: signalling the worker
     // alone could let the CPU fall asleep again before it runs.

app/src/main/java/com/raofflineproxy/service/ProxyService.kt

@@ -34,6 +34,7 @@ import com.raofflineproxy.data.CacheKeys
 import com.raofflineproxy.proxyUserAgent
 import com.raofflineproxy.proxy.AwardFlusher
 import com.raofflineproxy.proxy.CacheQueue
+import com.raofflineproxy.proxy.WatchFolders
 import com.raofflineproxy.proxy.RateLimitBackoff
 import com.raofflineproxy.proxy.GameActivity
 import com.raofflineproxy.proxy.ProxyServer
@@ -47,6 +48,7 @@ import com.raofflineproxy.proxy.DrainStop
 import com.raofflineproxy.update.AppUpdateNotifier
 import com.raofflineproxy.usage.UsageReporter
 import com.raofflineproxy.proxy.drainCacheQueue
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
@@ -89,6 +91,7 @@ class ProxyService : Service() {
     private var networkCallbackRegistered = false
     private var refreshJob: Job? = null
     private var cacheQueueJob: Job? = null
+    private var folderWatchJob: Job? = null
     private var cacheQueueRejectedToken: String? = null
     @Volatile private var queueLoginBlocked = false
         set(value) {
@@ -198,6 +201,10 @@ class ProxyService : Service() {
             cacheQueueJob = serviceScope.launch { cacheQueueLoop() }
         }
 
+        if (folderWatchJob?.isActive != true) {
+            folderWatchJob = serviceScope.launch { folderWatchLoop() }
+        }
+
         CachingNotifications.clearStandalone(this)
         if (cachingObserverJob?.isActive != true) {
             cachingObserverJob = serviceScope.launch {
@@ -360,6 +367,48 @@ class ProxyService : Service() {
         return (resumeAt - System.currentTimeMillis()).coerceAtLeast(1_000L)
     }
 
+    private suspend fun folderWatchLoop() {
+        while (true) {
+            try {
+                watchFoldersOnce()
+            } catch (c: CancellationException) {
+                throw c
+            } catch (t: Throwable) {
+                Log.e(TAG, "Watched folder round failed", t)
+            }
+            withTimeoutOrNull(CACHE_QUEUE_POLL_MS.milliseconds) { folderWatchWake.receive() }
+        }
+    }
+
+    /** Hashes and queues what watched folders gained since their last scan; the cache queue
+     *  then caches it within the budget. Runs as a bulk run so the queue worker doesn't drain
+     *  while it is still filling, and stands down whenever a game is being played. */
+    private suspend fun watchFoldersOnce() {
+        val folders = WatchFolders.due(db, System.currentTimeMillis())
+        if (folders.isEmpty() || !canWatchFolders()) return
+        val credentials = loadLoginCredentials(db) ?: return
+        if (credentials.token == cacheQueueRejectedToken) return
+        FolderWatchWakeLock.hold(this)
+        try {
+            CacheQueue.duringBulkRun {
+                for (folder in folders) {
+                    if (!canWatchFolders(ignoreBulkRun = true)) break
+                    val scan = WatchFolders.scan(this, db, folder, shouldAbort = { !canWatchFolders(ignoreBulkRun = true) })
+                    Log.i(
+                        TAG,
+                        "Watched folder ${folder.name}: " + (scan?.let { "${it.files} file(s), ${it.new} new, ${it.queued} queued" } ?: "unreadable")
+                    )
+                }
+            }
+        } finally {
+            FolderWatchWakeLock.release()
+            wakeCacheQueue()
+        }
+    }
+
+    private fun canWatchFolders(ignoreBulkRun: Boolean = false): Boolean =
+        (ignoreBulkRun || !CacheQueue.bulkRunActive) && isServerReachable() && onlineRefreshIdleDelayMs() <= 0
+
     private fun canWorkOnCacheQueue(): Boolean =
         !CacheQueue.bulkRunActive && isServerReachable() && onlineRefreshIdleDelayMs() <= 0
 
@@ -442,6 +491,7 @@ class ProxyService : Service() {
         }
         CacheQueueAlarm.cancel(this)
         CacheQueueWakeLock.release()
+        FolderWatchWakeLock.release()
         proxyServer.stop()
         if (networkCallbackRegistered) {
             runCatching { connectivityManager.unregisterNetworkCallback(networkCallback) }
@@ -679,6 +729,7 @@ class ProxyService : Service() {
         @Volatile
         private var runningInProcess = false
         private val cacheQueueWake = Channel<Unit>(Channel.CONFLATED)
+        private val folderWatchWake = Channel<Unit>(Channel.CONFLATED)
         private val runtimeLock = Any()
         private val _runtime = MutableStateFlow(ServiceRuntime())
 
@@ -695,6 +746,10 @@ class ProxyService : Service() {
             cacheQueueWake.trySend(Unit)
         }
 
+        fun wakeFolderWatch() {
+            folderWatchWake.trySend(Unit)
+        }
+
         private fun restartPendingIntent(context: Context): PendingIntent = PendingIntent.getBroadcast(
             context,
             RESTART_REQUEST_CODE,