feat(media): new file cache architecture, TTL for user-cache, new data settings ui

app/src/androidTest/java/org/monogram/CacheTtlRoomTest.kt

@@ -0,0 +1,133 @@
+package org.monogram.core.database
+
+import androidx.room.Room
+import androidx.test.ext.junit.runners.AndroidJUnit4
+import androidx.test.platform.app.InstrumentationRegistry
+import kotlinx.coroutines.runBlocking
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertFalse
+import org.junit.Assert.assertNotNull
+import org.junit.Assert.assertNull
+import org.junit.Assert.assertTrue
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.monogram.core.database.entity.MessageHoleEntity
+import org.monogram.core.database.entity.MessageTombstoneEntity
+import org.monogram.core.database.entity.ProfileCommonEntity
+import org.monogram.core.database.entity.ProfileMediaEntity
+import org.monogram.core.database.entity.ProfileMemberEntity
+import org.monogram.core.database.entity.ProfileTabsEntity
+import org.monogram.core.database.entity.UpdateCursorEntity
+import org.monogram.core.models.Chat
+import org.monogram.core.models.Message
+import org.monogram.core.models.MessageId
+import org.monogram.core.models.PeerId
+
+@RunWith(AndroidJUnit4::class)
+class CacheTtlRoomTest {
+    @Test
+    fun pruneKeepsTheRecentPageAndLeavesCursorsAlone() = runBlocking {
+        val context = InstrumentationRegistry.getInstrumentation().targetContext
+        val database = Room.inMemoryDatabaseBuilder(context, MonogramDatabase::class.java)
+            .allowMainThreadQueries()
+            .build()
+        val now = 1_800_000_000_000L
+        val old = now / 1000 - CACHE_MESSAGE_TTL_SECONDS - 60
+        val recent = now / 1000 - 3_600
+        val warmup = OfflineWarmup(database)
+        try {
+            val kept = (1..121).map { id ->
+                message(7, id, if (id == 121) recent else old)
+            }
+            val pending = message(7, -1, old, pending = true)
+            val failed = message(7, 200, old, failed = true)
+            val openChat = (1..30).map { id -> message(9, id, old) }
+            run {
+                database.messageDao()
+                    .upsertAll((kept + pending + failed + openChat).map { it.toEntity() })
+                database.messageHoleDao().insert(MessageHoleEntity(7L, 1, Int.MAX_VALUE))
+                database.messageTombstoneDao().insert(
+                    listOf(MessageTombstoneEntity(7L, 3), MessageTombstoneEntity(7L, 50)),
+                )
+                database.updateCursorDao().upsert(UpdateCursorEntity("updates", 4, 5, 6, 7, 0))
+                database.updateCursorDao().upsert(
+                    UpdateCursorEntity("$KEY_CHANNEL_CURSOR_PREFIX" + "1", 1, 0, 2, 0, 9),
+                )
+                database.profileTabsDao().upsert(ProfileTabsEntity(1L, "{}", 0L))
+                database.profileTabsDao().upsert(ProfileTabsEntity(2L, "{}", now))
+                database.profileMembersDao().upsertAll(
+                    listOf(
+                        ProfileMemberEntity(1L, 5L, 0, "Old", role = "member"),
+                        ProfileMemberEntity(2L, 6L, 0, "New", role = "member"),
+                    ),
+                )
+                database.profileMediaDao().upsertAll(
+                    listOf(
+                        ProfileMediaEntity(1L, "photo", 1, 0, 7L),
+                        ProfileMediaEntity(2L, "photo", 1, 0, 7L),
+                    ),
+                )
+                database.profileCommonDao().upsertAll(
+                    listOf(
+                        ProfileCommonEntity(1L, 3L, 0, "Old"),
+                        ProfileCommonEntity(2L, 4L, 0, "New"),
+                    ),
+                )
+            }
+            warmup.upsertChats(listOf(Chat(id = PeerId(7), title = "Kept")))
+            warmup.setOpenChat(9L)
+            warmup.cacheHistoryPage(PeerId(7), listOf(message(7, 121, recent)))
+            warmup.pruneExpiredCache(nowMillis = now, force = true)
+            run {
+                val dao = database.messageDao()
+                assertNull(dao.byIds(7L, listOf(1)).singleOrNull())
+                assertNotNull(dao.byIds(7L, listOf(22)).singleOrNull())
+                assertNotNull(dao.byIds(7L, listOf(121)).singleOrNull())
+                assertTrue(dao.byIds(7L, listOf(-1)).single().pending)
+                assertTrue(dao.byIds(7L, listOf(200)).single().failed)
+                assertEquals(30, dao.newestConfirmedIds(9L, 100).size)
+                assertEquals("Kept", database.chatDao().get(7L)?.title)
+                val account = database.updateCursorDao().get("updates")
+                assertEquals(4, account?.pts)
+                assertEquals(5, account?.qts)
+                assertEquals(6, account?.date)
+                assertEquals(7, account?.seq)
+                val channel = database.updateCursorDao().get("$KEY_CHANNEL_CURSOR_PREFIX" + "1")
+                assertEquals(9, channel?.channelPts)
+                assertNull(database.profileTabsDao().get(1L))
+                assertNotNull(database.profileTabsDao().get(2L))
+                assertTrue(database.profileMembersDao().list(1L).isEmpty())
+                assertEquals(1, database.profileMembersDao().list(2L).size)
+                assertTrue(database.profileMediaDao().list(1L).isEmpty())
+                assertEquals(1, database.profileMediaDao().list(2L).size)
+                assertTrue(database.profileCommonDao().list(1L).isEmpty())
+                assertEquals(1, database.profileCommonDao().list(2L).size)
+                assertTrue(database.messageTombstoneDao().deletedIds(7L, listOf(3)).isEmpty())
+                assertEquals(listOf(50), database.messageTombstoneDao().deletedIds(7L, listOf(50)))
+                val holes = database.messageHoleDao().forChat(7L, MessageHoleEntity.SOURCE_HISTORY)
+                assertTrue(holes.any { 1 in it.startId..it.endId })
+                assertTrue(holes.none { hole -> (22..121).any { it in hole.startId..hole.endId } })
+                assertFalse(holes.any { 121 in it.startId..it.endId })
+            }
+        } finally {
+            database.close()
+        }
+    }
+
+    private fun message(
+        chatId: Long,
+        id: Int,
+        date: Long,
+        pending: Boolean = false,
+        failed: Boolean = false,
+    ) = Message(
+        id = MessageId(PeerId(chatId), id),
+        senderId = null,
+        text = null,
+        date = date,
+        outgoing = pending || failed,
+        pending = pending,
+        failed = failed,
+        randomId = if (pending) 99L else null,
+    )
+}

app/src/androidTest/java/org/monogram/MediaStreamingTest.kt

@@ -14,10 +14,98 @@ import org.monogram.core.models.MessageId
 import org.monogram.core.models.PeerId
 import org.monogram.network.http.TelegramChunkFetcher
 import org.monogram.network.http.TelegramVideoDataSource
+import org.monogram.network.http.FileCache
 import java.io.File
 import java.io.RandomAccessFile
 
 class MediaStreamingTest {
+    @Test
+    fun streamedRangesSurviveClearUntilAllReadersClose() {
+        val context = InstrumentationRegistry.getInstrumentation().targetContext
+        val directory = File(context.cacheDir, "streamed-range-lease")
+        directory.deleteRecursively()
+        val cache = FileCache(directory)
+        val expected = ByteArray(TelegramVideoDataSource.PART_SIZE.toInt()) { (it % 251).toByte() }
+        val message = Message(
+            MessageId(PeerId(42), 12),
+            senderId = null,
+            text = null,
+            date = 0,
+            outgoing = false,
+            mediaKind = "video",
+            mediaCacheKey = "video:range-lease",
+            fileSize = expected.size.toLong()
+        )
+        var fetches = 0
+        val factory = TelegramVideoDataSource.Factory(
+            message,
+            directory,
+            TelegramChunkFetcher { _, _, path, _ ->
+                fetches++
+                File(path).writeBytes(expected)
+                Outcome.Ok(path)
+            },
+            cache
+        )
+        val first = factory.createDataSource()
+        val second = factory.createDataSource()
+        val spec = DataSpec.Builder().setUri("telegram://media/lease").build()
+        try {
+            first.open(spec)
+            second.open(spec)
+            assertEquals(1, first.read(ByteArray(1), 0, 1))
+            assertEquals(1, second.read(ByteArray(1), 0, 1))
+            assertEquals(1, fetches)
+            assertEquals(0L, cache.clear())
+            first.close()
+            val actual = ByteArray(32)
+            assertEquals(32, second.read(actual, 0, 32))
+            assertArrayEquals(expected.copyOfRange(1, 33), actual)
+            assertEquals(expected.size.toLong(), cache.totalBytes())
+            second.close()
+            assertEquals(0L, cache.totalBytes())
+        } finally {
+            first.close()
+            second.close()
+            directory.deleteRecursively()
+        }
+    }
+
+    @Test
+    fun cachedPlaybackKeepsLeaseUntilReaderCloses() = kotlinx.coroutines.runBlocking {
+        val context = InstrumentationRegistry.getInstrumentation().targetContext
+        val directory = File(context.cacheDir, "leased-playback-regression")
+        directory.deleteRecursively()
+        val repository = org.monogram.network.http.MediaRepository(
+            directory,
+            telegramFetcher = org.monogram.network.http.TelegramMediaFetcher { _, _, path, _, _ ->
+                File(path).writeBytes(byteArrayOf(1, 2, 3, 4))
+                Outcome.Ok(path)
+            })
+        val source = repository.createCachedDataSourceFactory("photo:lease").createDataSource()
+        try {
+            val result = repository.ensureIndexedMedia(
+                PeerId(42),
+                1,
+                "photo:lease",
+                priority = org.monogram.network.http.MediaPriority.USER
+            )
+            assertTrue(result is Outcome.Ok)
+            val file = (result as Outcome.Ok).value
+            source.open(DataSpec.Builder().setUri(Uri.fromFile(file)).build())
+            assertEquals(0L, repository.clearCache())
+            val bytes = ByteArray(4)
+            assertEquals(4, source.read(bytes, 0, 4))
+            assertArrayEquals(byteArrayOf(1, 2, 3, 4), bytes)
+            source.close()
+            assertTrue(!file.exists())
+        } finally {
+            source.close()
+            repository.shutdown()
+            directory.deleteRecursively()
+        }
+    }
+
     @Test
     fun arbitrarySeekReadsAlignedPartsAndCleansTemporaryFiles() {
         val context = InstrumentationRegistry.getInstrumentation().targetContext
@@ -34,7 +122,7 @@ class MediaStreamingTest {
             val start = offset.toInt()
             val chunk = bytes.copyOfRange(start, minOf(start + partSize, bytes.size))
             RandomAccessFile(destination, "rw").use { file ->
-                file.seek(offset)
+                file.setLength(0)
                 file.write(chunk)
             }
             Outcome.Ok(destination)
@@ -83,7 +171,7 @@ class MediaStreamingTest {
             TelegramChunkFetcher { _, _, destination, requested ->
                 assertEquals(offset, requested)
                 RandomAccessFile(destination, "rw").use { file ->
-                    file.seek(requested)
+                    file.setLength(0)
                     file.write(partial)
                 }
                 Outcome.Ok(destination)