package computer.whatthefuck.arcology.indexer

import com.github.ajalt.clikt.core.CliktCommand
import com.github.ajalt.clikt.core.CliktError
import com.github.ajalt.clikt.parameters.options.default
import com.github.ajalt.clikt.parameters.options.option
import com.github.ajalt.clikt.parameters.types.int
import com.github.ajalt.clikt.parameters.types.long
import kotlinx.coroutines.runBlocking
import java.io.File

class SyncCommand : CliktCommand(
    name = "sync",
    help = "Watch Syncthing for org file changes and keep the index up to date"
) {
    private val dbPath: String by option("--db", help = "Path to SQLite database").default("~/org/arcology.db")
    private val orgDir: String by option("--org-dir", help = "Root org-mode directory").default("~/org")
    private val apiUrl: String by option("--api-url", help = "Syncthing REST API base URL").default("http://127.0.0.1:8384")
    private val apiKeyOption: String? by option("--api-key", help = "Syncthing REST API key (defaults to \$ARCOLOGY_SYNCTHING_API_KEY)")
    private val folderId: String? by option("--folder", help = "Explicit Syncthing folder ID (overrides path matching)")
    private val pollTimeoutSeconds: Int by option("--poll-timeout", help = "Event long-poll timeout in seconds").int().default(60)
    private val rescanIntervalSeconds: Long by option(
        "--rescan-interval",
        help = "Run a reconciliation index pass after this many idle seconds (empty polls x poll timeout); 0 disables"
    ).long().default(3600)

    override fun run() {
        val apiKey = apiKeyOption ?: System.getenv("ARCOLOGY_SYNCTHING_API_KEY")
        if (apiKey.isNullOrBlank()) {
            throw CliktError("--api-key or ARCOLOGY_SYNCTHING_API_KEY must be set")
        }
        val expandedDb = dbPath.replace("~", System.getProperty("user.home"))
        val expandedOrgDir = orgDir.replace("~", System.getProperty("user.home"))
        val orgPath = File(expandedOrgDir).absolutePath

        println("Starting Syncthing watch...")
        println("  Database: $expandedDb")
        println("  Org dir: $orgPath")
        println("  Syncthing API: $apiUrl")
        println("  Folder: ${folderId ?: "(resolved from --org-dir)"}")

        val indexer = createIndexingService(expandedDb, IndexingConfig(), orgPath)
        val client = KtorSyncthingClient(apiUrl, apiKey, pollTimeoutSeconds)
        val service = SyncthingWatchService(
            client = client,
            indexer = indexer,
            orgDir = orgPath,
            folderId = folderId,
            pollTimeoutSeconds = pollTimeoutSeconds,
            rescanIntervalSeconds = rescanIntervalSeconds,
        )
        runBlocking { service.watch() }
    }
}

package computer.whatthefuck.arcology.indexer

import kotlinx.coroutines.delay

const val SYNCTHING_LOCAL_CHANGE = "LocalChangeDetected"
const val SYNCTHING_REMOTE_CHANGE = "RemoteChangeDetected"

<<watcher-helpers>>

/**
 * Watch one Syncthing folder and keep the index current with the files
 * on disk. Call [startup] once, then run [watch]; it long-polls the event
 * stream forever (or for [watch]'s [maxIterations] polls, for tests) and
 * feeds each change to the indexer's single-file path, which runs all
 * IndexerPlugins and skips unchanged content by hash. When the stream
 * stays quiet for [rescanIntervalSeconds] of poll time, a reconciliation
 * index pass runs instead — a stall breaker that bounds index staleness
 * even if the event stream ever fails silently again.
 */
class SyncthingWatchService(
    private val client: SyncthingClient,
    private val indexer: FileIndexingService,
    private val orgDir: String,
    private val folderId: String? = null,
    private val pollTimeoutSeconds: Int = 60,
    private val rescanIntervalSeconds: Long = 3600,
    private val initialBackoffMs: Long = 10_000,
    private val maxBackoffMs: Long = 300_000,
    private val log: (String) -> Unit = { println(it) },
) {
    private var cursor: Long = 0L

    fun currentCursor(): Long = cursor

    <<watcher-service-methods>>
}

/**
 * Resolve the folder, run one full indexing pass, then seed the event
 * cursor so only future events are processed.
 */
suspend fun startup(): SyncthingFolder {
    val folder = resolveSyncthingFolder(client.folders(), orgDir, folderId)
    log("Watching Syncthing folder ${folder.id} (${folder.label ?: "unlabeled"}) at ${folder.path}")
    val result = indexer.indexDirectory(orgDir)
    log("Startup index: ${result.successful} indexed, ${result.skipped} unchanged, ${result.failed} failed")
    cursor = client.latestEventId() ?: 0L
    log("Watching for events after id $cursor")
    return folder
}

/**
 * Run the poll loop. Each iteration is one long-poll; transient failures
 * back off and retry with the cursor intact, and auth failures abort.
 * Empty batches are expected when the long-poll times out quietly — they
 * log every 15th poll as a heartbeat so a silent-but-healthy loop is
 * distinguishable from a stuck one. After ~[rescanIntervalSeconds] of
 * silence a reconciliation index pass runs (the stall breaker). No
 * ID-gap detection is attempted — the startup scan is the safety net,
 * and a restarted watcher rescans on startup.
 */
suspend fun watch(maxIterations: Int = Int.MAX_VALUE) {
    val folder = startup()
    var backoff = initialBackoffMs
    var emptyPolls = 0L
    // The long-poll blocks ~pollTimeoutSeconds when quiet, so wall-clock
    // idleness is approximated by counting empty polls.
    val rescanAfterPolls = if (rescanIntervalSeconds > 0) {
        (rescanIntervalSeconds / pollTimeoutSeconds.coerceAtLeast(1)).coerceAtLeast(1)
    } else {
        Long.MAX_VALUE
    }
    repeat(maxIterations) {
        val events = try {
            client.events(cursor, pollTimeoutSeconds)
        } catch (e: SyncthingAuthException) {
            throw SyncthingAuthException("Syncthing rejected the API key: ${e.message}", e)
        } catch (e: Exception) {
            log("Syncthing API unreachable (${e.message}); retrying in ${backoff / 1000}s")
            delay(backoff)
            backoff = (backoff * 2).coerceAtMost(maxBackoffMs)
            return@repeat
        }
        backoff = initialBackoffMs
        if (events.isEmpty()) {
            emptyPolls++
            if (emptyPolls % 15L == 0L) {
                log("No disk events for $emptyPolls polls; cursor=$cursor; continuing")
            }
            if (emptyPolls >= rescanAfterPolls) {
                emptyPolls = 0
                log("Idle for ~$rescanIntervalSeconds s; running reconciliation index pass")
                try {
                    val result = indexer.indexDirectory(orgDir)
                    log(
                        "Reconciliation index: ${result.successful} indexed, " +
                            "${result.skipped} unchanged, ${result.failed} failed"
                    )
                } catch (e: Exception) {
                    log("Reconciliation index failed: ${e.message}")
                }
            }
            return@repeat
        }
        emptyPolls = 0
        log("Got ${events.size} disk event(s) for folder ${folder.id}")
        processBatch(events, folder)
        cursor = events.last().id
    }
}

/**
 * Turn one batch of disk events into indexer calls. Both event types are
 * deduped per path with delete-wins semantics; only .org files reach the
 * indexer and directories are skipped. Item paths are relativized against
 * the folder root so event-path writes use the same file keys as the
 * directory pass; absolute paths outside the folder root are skipped with
 * a log line. Exceptions from a single batch are caught and logged so one
 * bad batch or indexing hiccup cannot kill the poll loop; the cursor
 * still advances past the batch.
 */
suspend fun processBatch(events: List<SyncthingEvent>, folder: SyncthingFolder) {
    val actions = LinkedHashMap<String, String>()
    for (event in events) {
        if (event.type != SYNCTHING_LOCAL_CHANGE && event.type != SYNCTHING_REMOTE_CHANGE) continue
        val data = event.data ?: continue
        val path = data.path ?: continue
        if (folder.id.isNotBlank() && data.folder != null && data.folder != folder.id) continue
        if (data.itemType != null && data.itemType != "file") continue
        val existing = actions[path]
        actions[path] = when {
            data.action == "deleted" -> "delete"
            existing == "delete" -> "delete"
            else -> "update"
        }
    }
    for ((item, action) in actions) {
        if (!isIndexableOrgItem(item)) continue
        if (isSyncConflictItem(item)) {
            log("Conflict file detected, indexing as-is (resolve it manually): $item")
        }
        val resolved = resolveSyncthingItemPath(item, folder.path)
        if (resolved.isEmpty()) {
            log("Skipping event outside folder root: $item")
            continue
        }
        try {
            when (action) {
                "delete" -> {
                    indexer.removeFile(resolved)
                    log("Removed from index: $resolved")
                }
                else -> {
                    when (val result = indexer.indexFile(resolved)) {
                        is FileIndexResult.Success ->
                            log("Indexed ${result.filePath} (${result.nodesCount} nodes)")
                        is FileIndexResult.Error ->
                            log("Index error for $resolved: ${result.message}")
                        is FileIndexResult.Skipped ->
                            log("Unchanged, skipped: $resolved")
                    }
                }
            }
        } catch (e: Exception) {
            log("Index dispatch error for $resolved: ${e.message}")
        }
    }
}

/**
 * Pick the Syncthing folder that backs [orgDir]. An explicit [folderId]
 * wins; otherwise any folder whose path equals orgDir or contains it
 * matches, and the most specific (longest) path wins when folders are
 * nested inside each other.
 */
fun resolveSyncthingFolder(
    folders: List<SyncthingFolder>,
    orgDir: String,
    folderId: String? = null,
): SyncthingFolder {
    if (folderId != null) {
        return folders.firstOrNull { it.id.equals(folderId, ignoreCase = true) }
            ?: throw SyncthingClientException(
                "No Syncthing folder with id '$folderId' (have: ${folders.map { it.id }})"
            )
    }
    val normalized = orgDir.trimEnd('/')
    val matches = folders.filter { folder ->
        val path = folder.path.trimEnd('/')
        path == normalized || normalized.startsWith("$path/")
    }
    if (matches.isEmpty()) {
        throw SyncthingClientException(
            "No Syncthing folder contains $orgDir; folder paths are: ${folders.map { it.path }}"
        )
    }
    return matches.maxBy { it.path.length }
}

/**
 ,* True for .org files the indexer should see. Syncthing noise — dotfiles,
 ,* .stfolder, .syncthing.*.tmp scratch files — is filtered here.
 ,*/
fun isIndexableOrgItem(item: String): Boolean {
    if (!item.lowercase().endsWith(".org")) return false
    return item.split('/').none { it.startsWith(".") }
}

fun isSyncConflictItem(item: String): Boolean =
    item.substringAfterLast('/').contains(".sync-conflict-")

fun resolveSyncthingItemPath(item: String, folderPath: String): String {
    if (!item.startsWith("/")) return item
    // An absolute item path must be relativized against the folder root so
    // the event path writes the SAME file keys the directory pass writes.
    // Feeding the absolute path to the indexer keyed rows under the host
    // mount (/media/org/...) while routes and the startup pass use keys
    // relative to the org root, which blanked every routed page the moment
    // an event-path write rewrote the node row's file column in place.
    val normalizedFolder = folderPath.trimEnd('/')
    if (item == normalizedFolder) return ""
    if (item.startsWith("$normalizedFolder/")) return item.removePrefix("$normalizedFolder/")
    // Item lives outside the folder root: skip it rather than writing an
    // orphan key; the caller logs when null comes back.
    return ""
}

package computer.whatthefuck.arcology.indexer

import kotlinx.coroutines.flow.emptyFlow
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.test.runTest
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertFailsWith
import kotlin.test.assertFalse
import kotlin.test.assertTrue

private class FakeSyncthingClient(
    var foldersList: List<SyncthingFolder> = emptyList(),
) : SyncthingClient {
    private sealed interface Step
    private class EmitStep(val events: List<SyncthingEvent>) : Step
    private class FailStep(val error: Exception) : Step

    private val steps = ArrayDeque<Step>()
    val eventCalls = mutableListOf<Pair<Long, Int>>()

    /**
     * Models the server's per-mask subscription: latestEventId() reports
     * the head of the SAME counter the events() poll reads (the disk
     * subscription), not some other mask's counter.
     */
    var seededEventId: Long? = null
    var latestEventIdCalls = 0

    fun emit(vararg events: SyncthingEvent) {
        steps.addLast(EmitStep(events.toList()))
    }

    fun fail(error: Exception) {
        steps.addLast(FailStep(error))
    }

    override suspend fun folders(): List<SyncthingFolder> = foldersList

    override suspend fun events(since: Long, timeoutSeconds: Int): List<SyncthingEvent> {
        eventCalls.add(since to timeoutSeconds)
        return when (val step = steps.removeFirstOrNull()) {
            is EmitStep -> step.events
            is FailStep -> throw step.error
            null -> emptyList()
        }
    }

    override suspend fun latestEventId(): Long? {
        latestEventIdCalls++
        return seededEventId
    }
}

private class RecordingIndexer : FileIndexingService {
    val indexed = mutableListOf<String>()
    val removed = mutableListOf<String>()
    var directoryPasses = 0
    var throwOnIndex: Exception? = null

    override suspend fun indexFile(filePath: String): FileIndexResult {
        throwOnIndex?.let { throw it }
        indexed.add(filePath)
        return FileIndexResult.Success(filePath, nodesCount = 1, linksCount = 0, tagsCount = 0)
    }

    override suspend fun removeFile(filePath: String): Boolean {
        removed.add(filePath)
        return true
    }

    override suspend fun indexDirectory(path: String, recursive: Boolean): IndexResult {
        directoryPasses++
        return IndexResult(totalFiles = 0, successful = 0, failed = 0, skipped = 0)
    }

    override fun indexDirectoryFlow(path: String, recursive: Boolean): Flow<IndexProgress> =
        emptyFlow()
}

private fun diskEvent(
    id: Long,
    path: String,
    action: String = "modified",
    folder: String? = "abc-def",
    itemType: String? = "file",
    modifiedBy: String? = null,
) = SyncthingEvent(
    id = id,
    type = if (modifiedBy == null) SYNCTHING_LOCAL_CHANGE else SYNCTHING_REMOTE_CHANGE,
    data = SyncthingEventData(
        path = path,
        folder = folder,
        action = action,
        itemType = itemType,
        modifiedBy = modifiedBy,
    ),
)

private fun watchService(
    client: SyncthingClient,
    indexer: FileIndexingService,
    orgDir: String = "/home/rrix/org",
    pollTimeoutSeconds: Int = 60,
    rescanIntervalSeconds: Long = 0,
) = SyncthingWatchService(
    client = client,
    indexer = indexer,
    orgDir = orgDir,
    pollTimeoutSeconds = pollTimeoutSeconds,
    rescanIntervalSeconds = rescanIntervalSeconds,
    initialBackoffMs = 1,
    maxBackoffMs = 4,
    log = {},
)

class SyncthingWatchServiceTest {

    @Test
    fun `resolveFolder matches exact path`() {
        val folders = listOf(
            SyncthingFolder("abc-def", "Org", "/home/rrix/org"),
            SyncthingFolder("other", "Other", "/home/other"),
        )
        assertEquals("abc-def", resolveSyncthingFolder(folders, "/home/rrix/org").id)
    }

    @Test
    fun `resolveFolder matches folder containing orgDir`() {
        val folders = listOf(SyncthingFolder("wide", null, "/home/rrix"))
        assertEquals("wide", resolveSyncthingFolder(folders, "/home/rrix/org").id)
    }

    @Test
    fun `resolveFolder prefers most specific nested folder`() {
        val folders = listOf(
            SyncthingFolder("wide", null, "/home/rrix"),
            SyncthingFolder("specific", null, "/home/rrix/org"),
        )
        assertEquals("specific", resolveSyncthingFolder(folders, "/home/rrix/org").id)
    }

    @Test
    fun `resolveFolder explicit id overrides path matching`() {
        val folders = listOf(SyncthingFolder("abc-def", "Org", "/somewhere/else"))
        assertEquals("abc-def", resolveSyncthingFolder(folders, "/home/rrix/org", folderId = "abc-def").id)
    }

    @Test
    fun `resolveFolder throws when nothing matches`() {
        val folders = listOf(SyncthingFolder("other", null, "/home/other"))
        assertFailsWith<SyncthingClientException> {
            resolveSyncthingFolder(folders, "/home/rrix/org")
        }
    }

    @Test
    fun `resolveFolder throws for unknown explicit id`() {
        assertFailsWith<SyncthingClientException> {
            resolveSyncthingFolder(emptyList(), "/home/rrix/org", folderId = "nope")
        }
    }

    @Test
    fun `indexable items are org files without dotfile segments`() {
        assertTrue(isIndexableOrgItem("journals/2026/journal.org"))
        assertTrue(isIndexableOrgItem("index.org"))
        assertFalse(isIndexableOrgItem("notes.md"))
        assertFalse(isIndexableOrgItem(".stfolder"))
        assertFalse(isIndexableOrgItem(".stfolder/notes.org"))
        assertFalse(isIndexableOrgItem(".syncthing.journal.org.tmp"))
        assertFalse(isIndexableOrgItem("journals/.hidden.org"))
    }

    @Test
    fun `conflict files are detected`() {
        assertTrue(isSyncConflictItem("journals/foo.sync-conflict-20260909-120000.org"))
        assertFalse(isSyncConflictItem("journals/foo.org"))
    }

    @Test
    fun `item paths relativize against the folder root`() {
        // The event path MUST write the same file keys as the directory
        // pass, which are relative to the org root.
        assertEquals("the_lions_rear.org", resolveSyncthingItemPath("the_lions_rear.org", "/media/org"))
        assertEquals("journals/j.org", resolveSyncthingItemPath("journals/j.org", "/media/org"))
        assertEquals("the_lions_rear.org", resolveSyncthingItemPath("/media/org/the_lions_rear.org", "/media/org"))
        assertEquals("a/b.org", resolveSyncthingItemPath("/media/org/a/b.org", "/media/org/"))
        // Outside the folder root: empty string means "skip" to the caller.
        assertEquals("", resolveSyncthingItemPath("/elsewhere/a.org", "/media/org"))
        assertEquals("", resolveSyncthingItemPath("/media/org", "/media/org"))
        // The old behavior — absolute passthrough — is the double-keyspace
        // regression this helper guards against.
        assertTrue(!resolveSyncthingItemPath("/media/org/a.org", "/media/org").startsWith("/"))
    }

    @Test
    fun `startup resolves folder, indexes directory, seeds cursor`() = runTest {
        val client = FakeSyncthingClient(
            listOf(SyncthingFolder("abc-def", "Org", "/home/rrix/org"))
        ).apply { seededEventId = 42 }
        val indexer = RecordingIndexer()
        val service = watchService(client, indexer)

        val folder = service.startup()

        assertEquals("abc-def", folder.id)
        assertEquals(1, indexer.directoryPasses)
        assertEquals(42, service.currentCursor())
    }

    @Test
    fun `processBatch indexes updates and keys rows relative to org root`() = runTest {
        val client = FakeSyncthingClient()
        val indexer = RecordingIndexer()
        val service = watchService(client, indexer)
        val folder = SyncthingFolder("abc-def", null, "/home/rrix/org")

        // A real LocalChangeDetected payload carries an absolute path; the
        // double-keyspace incident wrote rows keyed by this absolute path
        // while routes and the startup pass used root-relative keys, which
        // blanked routed pages. The event path must relativize.
        service.processBatch(
            listOf(
                diskEvent(1, "/home/rrix/org/journals/j.org"),
                diskEvent(2, "notes.md"),
                diskEvent(3, "/home/rrix/org/images/pic.png"),
                diskEvent(4, "/elsewhere/outside.org"),
            ),
            folder,
        )

        assertEquals(listOf("journals/j.org"), indexer.indexed)
        assertTrue(indexer.removed.isEmpty())
    }

    @Test
    fun `processBatch relativizes absolute item paths and skips outside-root items`() = runTest {
        val client = FakeSyncthingClient()
        val logs = mutableListOf<String>()
        val indexer = RecordingIndexer()
        val service = SyncthingWatchService(
            client = client,
            indexer = indexer,
            orgDir = "/home/rrix/org",
            pollTimeoutSeconds = 60,
            initialBackoffMs = 1,
            maxBackoffMs = 4,
            log = { logs.add(it) },
        )
        val folder = SyncthingFolder("abc-def", null, "/media/org")

        service.processBatch(
            listOf(
                diskEvent(1, "/media/org/the_lions_rear.org"),
                diskEvent(2, "/media/org/nested/sub.org"),
                diskEvent(3, "/not-under-root/rogue.org"),
            ),
            folder,
        )

        assertEquals(
            listOf("the_lions_rear.org", "nested/sub.org"),
            indexer.indexed,
        )
        assertTrue(logs.any { it.contains("Skipping event outside folder root: /not-under-root/rogue.org") })
    }

    @Test
    fun `processBatch dispatches deletes to removeFile`() = runTest {
        val client = FakeSyncthingClient()
        val indexer = RecordingIndexer()
        val service = watchService(client, indexer)
        val folder = SyncthingFolder("abc-def", null, "/home/rrix/org")

        service.processBatch(
            listOf(
                diskEvent(1, "journals/j.org", action = "deleted"),
            ),
            folder,
        )

        assertEquals(listOf("journals/j.org"), indexer.removed)
        assertTrue(indexer.indexed.isEmpty())
    }

    @Test
    fun `processBatch dedupes with delete-wins`() = runTest {
        val client = FakeSyncthingClient()
        val indexer = RecordingIndexer()
        val service = watchService(client, indexer)
        val folder = SyncthingFolder("abc-def", null, "/home/rrix/org")

        service.processBatch(
            listOf(
                diskEvent(1, "journals/j.org"),
                diskEvent(2, "journals/j.org", action = "deleted"),
            ),
            folder,
        )

        assertEquals(listOf("journals/j.org"), indexer.removed)
        assertTrue(indexer.indexed.isEmpty())
    }

    @Test
    fun `processBatch ignores other folders, errors, and directories`() = runTest {
        val client = FakeSyncthingClient()
        val indexer = RecordingIndexer()
        val service = watchService(client, indexer)
        val folder = SyncthingFolder("abc-def", null, "/home/rrix/org")

        service.processBatch(
            listOf(
                diskEvent(1, "j.org", folder = "other-folder"),
                diskEvent(2, "subdir", itemType = "dir"),
                diskEvent(3, "keep.org", modifiedBy = "BPDFDTU"),
            ),
            folder,
        )

        assertEquals(listOf("keep.org"), indexer.indexed)
    }

    @Test
    fun `watch processes batches and advances cursor`() = runTest {
        val client = FakeSyncthingClient(
            listOf(SyncthingFolder("abc-def", null, "/home/rrix/org"))
        ).apply {
            seededEventId = 0
            emit(diskEvent(1, "a.org"))
            emit()
        }
        val indexer = RecordingIndexer()
        val service = watchService(client, indexer)

        service.watch(maxIterations = 2)

        assertEquals(listOf("a.org"), indexer.indexed)
        assertEquals(1, service.currentCursor())
        assertEquals(1, indexer.directoryPasses)
    }

    @Test
    fun `watch survives event buffer rotation with sparse ids`() = runTest {
        // One subscription's buffer holds 1000 events per mask; ids belong
        // to that subscription's counter, so a batch spanning rotated ids
        // is still processed in full and the cursor lands on the last one.
        val client = FakeSyncthingClient(
            listOf(SyncthingFolder("abc-def", null, "/home/rrix/org"))
        ).apply {
            seededEventId = 0
            emit(diskEvent(1001, "a.org"), diskEvent(1042, "b.org"))
        }
        val indexer = RecordingIndexer()
        val service = watchService(client, indexer)

        service.watch(maxIterations = 1)

        assertEquals(1, indexer.directoryPasses)
        assertEquals(1042, service.currentCursor())
        assertEquals(
            listOf("a.org", "b.org"),
            indexer.indexed
        )
    }

    @Test
    fun `watch seeds cursor in disk id space and still sees low ids`() = runTest {
        // Regression for the silent-stall incident: the seed reported id
        // 500 (a value from a foreign subscription's counter) while disk
        // events carry low ids 1..
        // Before the fix the poll loop polled since=500 against disk ids 1, 2
        // and starved forever; after the fix the seed and the poll read the
        // same subscription, so both events are indexed.
        val client = FakeSyncthingClient(
            listOf(SyncthingFolder("abc-def", null, "/home/rrix/org"))
        ).apply {
            seededEventId = 500
            emit(diskEvent(1, "a.org"))
            emit(diskEvent(2, "b.org"))
        }
        val indexer = RecordingIndexer()
        val service = watchService(client, indexer)

        service.watch(maxIterations = 2)

        assertEquals(
            listOf("a.org", "b.org"),
            indexer.indexed
        )
        assertEquals(2, service.currentCursor())
        assertEquals(1, indexer.directoryPasses)
        assertEquals(1, client.latestEventIdCalls)
    }

    @Test
    fun `watch logs heartbeat on repeated empty batches`() = runTest {
        val logs = mutableListOf<String>()
        val client = FakeSyncthingClient(
            listOf(SyncthingFolder("abc-def", null, "/home/rrix/org"))
        ).apply { seededEventId = 0 }
        val indexer = RecordingIndexer()
        val service = SyncthingWatchService(
            client = client,
            indexer = indexer,
            orgDir = "/home/rrix/org",
            pollTimeoutSeconds = 60,
            initialBackoffMs = 1,
            maxBackoffMs = 4,
            log = { logs.add(it) },
        )

        service.watch(maxIterations = 20)

        assertTrue(indexer.indexed.isEmpty())
        assertEquals(20, client.eventCalls.size)
        assertTrue(logs.any { it.contains("No disk events for 15 polls") })
    }

    @Test
    fun `watch runs reconciliation pass after idle interval`() = runTest {
        // 30s interval against a 10s poll window -> rescan after 3 empty
        // polls. directoryPasses counts the startup pass plus the
        // reconciliation pass; no heartbeat-triggered duplicates within 5
        // polls because the idle counter resets after each pass.
        val logs = mutableListOf<String>()
        val client = FakeSyncthingClient(
            listOf(SyncthingFolder("abc-def", null, "/home/rrix/org"))
        ).apply { seededEventId = 0 }
        val indexer = RecordingIndexer()
        val service = SyncthingWatchService(
            client = client,
            indexer = indexer,
            orgDir = "/home/rrix/org",
            pollTimeoutSeconds = 10,
            rescanIntervalSeconds = 30,
            initialBackoffMs = 1,
            maxBackoffMs = 4,
            log = { logs.add(it) },
        )

        service.watch(maxIterations = 5)

        assertEquals(2, indexer.directoryPasses) // startup + one reconciliation
        assertTrue(logs.any { it.contains("Idle for ~30 s; running reconciliation index pass") })
        assertTrue(logs.any { it.contains("Reconciliation index: 0 indexed, 0 unchanged, 0 failed") })
    }

    @Test
    fun `rescan disabled by default in test factory`() = runTest {
        // rescanIntervalSeconds defaults to 0 in the factory: a quiet
        // watcher never rescans, so 25 empty polls trigger no directory
        // pass beyond the startup one.
        val client = FakeSyncthingClient(
            listOf(SyncthingFolder("abc-def", null, "/home/rrix/org"))
        ).apply { seededEventId = 0 }
        val indexer = RecordingIndexer()
        val service = watchService(client, indexer)

        service.watch(maxIterations = 25)

        assertEquals(1, indexer.directoryPasses) // startup only
    }

    @Test
    fun `watch keeps running when indexing one item throws`() = runTest {
        val client = FakeSyncthingClient(
            listOf(SyncthingFolder("abc-def", null, "/home/rrix/org"))
        ).apply {
            seededEventId = 0
            emit(diskEvent(1, "a.org"))
            emit(diskEvent(2, "b.org"))
        }
        val indexer = RecordingIndexer().apply {
            throwOnIndex = IllegalStateException("disk exploded")
        }
        val logs = mutableListOf<String>()
        val service = SyncthingWatchService(
            client = client,
            indexer = indexer,
            orgDir = "/home/rrix/org",
            pollTimeoutSeconds = 60,
            initialBackoffMs = 1,
            maxBackoffMs = 4,
            log = { logs.add(it) },
        )

        service.watch(maxIterations = 2)

        assertEquals(2, service.currentCursor())
        assertTrue(logs.any { it.contains("Index dispatch error") && it.contains("disk exploded") })
        assertTrue(logs.any { it.contains("Got 1 disk event(s)") })
    }

    @Test
    fun `watch survives transient failures with backoff`() = runTest {
        val client = FakeSyncthingClient(
            listOf(SyncthingFolder("abc-def", null, "/home/rrix/org"))
        ).apply {
            seededEventId = 0
            fail(RuntimeException("connection refused"))
            emit(diskEvent(1, "a.org"))
        }
        val indexer = RecordingIndexer()
        val service = watchService(client, indexer)

        service.watch(maxIterations = 2)

        assertEquals(listOf("a.org"), indexer.indexed)
        assertEquals(1, service.currentCursor())
    }

    @Test
    fun `watch aborts on auth failure`() = runTest {
        val client = FakeSyncthingClient(
            listOf(SyncthingFolder("abc-def", null, "/home/rrix/org"))
        ).apply { seededEventId = 0; fail(SyncthingAuthException("403")) }
        val indexer = RecordingIndexer()
        val service = watchService(client, indexer)

        assertFailsWith<SyncthingAuthException> {
            service.watch(maxIterations = 1)
        }
    }
}

package computer.whatthefuck.arcology.indexer

import kotlinx.serialization.SerialName
import kotlinx.serialization.Serializable

/**
 * A folder Syncthing shares, as reported by /rest/config/folders. Only the
 * fields the watcher needs are modeled; the DTOs tolerate unknown keys so
 * newer Syncthing versions don't break parsing.
 */
@Serializable
data class SyncthingFolder(
    val id: String,
    val label: String? = null,
    val path: String,
)

@Serializable
data class SyncthingEvent(
    val id: Long,
    @SerialName("globalID") val globalId: Long? = null,
    val time: String? = null,
    val type: String,
    val data: SyncthingEventData? = null,
)

@Serializable
data class SyncthingEventData(
    val path: String? = null,
    val folder: String? = null,
    val action: String? = null,
    @SerialName("modifiedBy") val modifiedBy: String? = null,
    @SerialName("type") val itemType: String? = null,
)

/**
 * Thrown for transport-level failures the watch loop can retry.
 */
open class SyncthingClientException(message: String, cause: Throwable? = null) :
    Exception(message, cause)

/**
 * Thrown when Syncthing rejects the API key. The watch loop treats this as
 * fatal: retrying with the same key cannot succeed.
 */
class SyncthingAuthException(message: String, cause: Throwable? = null) :
    SyncthingClientException(message, cause)

/**
 * Minimal interface over the Syncthing REST API so the watch service stays
 * platform-independent. Platform implementations live in jvmMain (Ktor) and,
 * eventually, androidMain (OkHttp).
 */
interface SyncthingClient {
    suspend fun folders(): List<SyncthingFolder>

    /**
     * Long-poll the event stream. Pass [timeoutSeconds] as the server-side
     * long-poll window; returns an empty list if it expires with no events.
     */
    suspend fun events(since: Long, timeoutSeconds: Int): List<SyncthingEvent>

    /**
     * The ID of the most recent buffered event, or null if none exist.
     * Implementations MUST use the same event mask for seeding as for
     * polling: REST event ids are per-subscription, so a cursor taken from
     * a differently-filtered subscription belongs to a different ID space
     * and makes every subsequent poll miss or stall until the polled
     * subscription's counter climbs past it.
     */
    suspend fun latestEventId(): Long?
}

package computer.whatthefuck.arcology.indexer

import io.ktor.client.HttpClient
import io.ktor.client.HttpClientConfig
import io.ktor.client.engine.HttpClientEngine
import io.ktor.client.engine.cio.CIO
import io.ktor.client.plugins.ClientRequestException
import io.ktor.client.plugins.HttpTimeout
import io.ktor.client.request.HttpRequestBuilder
import io.ktor.client.request.get
import io.ktor.client.request.header
import io.ktor.client.request.parameter
import io.ktor.client.statement.bodyAsText
import kotlinx.serialization.json.Json

class KtorSyncthingClient(
    private val baseUrl: String,
    private val apiKey: String,
    pollTimeoutSeconds: Int = 60,
    engine: HttpClientEngine? = null,
) : SyncthingClient {

    private val json = Json { ignoreUnknownKeys = true }

    private val clientConfig: HttpClientConfig<*>.() -> Unit = {
        expectSuccess = true
        install(HttpTimeout) {
            requestTimeoutMillis = (pollTimeoutSeconds + 30) * 1000L
        }
    }

    // Tests inject a MockEngine; production takes the CIO factory.
    private val client = if (engine != null) {
        HttpClient(engine, clientConfig)
    } else {
        HttpClient(CIO, clientConfig)
    }

    private suspend fun request(
        path: String,
        configure: HttpRequestBuilder.() -> Unit = {},
    ): String =
        try {
            client.get("$baseUrl$path") {
                header("X-API-Key", apiKey)
                configure()
            }.bodyAsText()
        } catch (e: ClientRequestException) {
            if (e.response.status.value == 403) {
                throw SyncthingAuthException("Syncthing returned 403 — check the API key", e)
            }
            throw SyncthingClientException("Syncthing request to $path failed: ${e.message}", e)
        }

    override suspend fun folders(): List<SyncthingFolder> =
        json.decodeFromString<List<SyncthingFolder>>(request("/rest/config/folders"))

    override suspend fun events(since: Long, timeoutSeconds: Int): List<SyncthingEvent> =
        json.decodeFromString<List<SyncthingEvent>>(
            request("/rest/events") {
                parameter("events", "$SYNCTHING_LOCAL_CHANGE,$SYNCTHING_REMOTE_CHANGE")
                parameter("since", since)
                parameter("timeout", timeoutSeconds)
            }
        )

    override suspend fun latestEventId(): Long? =
        json.decodeFromString<List<SyncthingEvent>>(
            request("/rest/events") {
                // The cursor must come from the same subscription the poll
                // loop reads: /rest/events caches one subscription per event
                // mask and ids are per-subscription, so seeding without the
                // disk-event filter yields ids from a different ID space and
                // the poll loop stalls (see the narrative above).
                parameter("events", "$SYNCTHING_LOCAL_CHANGE,$SYNCTHING_REMOTE_CHANGE")
                parameter("since", 0)
                parameter("limit", 1)
            }
        ).lastOrNull()?.id
}

package computer.whatthefuck.arcology.indexer

import io.ktor.client.engine.mock.MockEngine
import io.ktor.client.engine.mock.respondOk
import io.ktor.http.*
import kotlinx.coroutines.test.runTest
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertTrue

class KtorSyncthingClientTest {

    private fun diskEventsJson(ids: List<Long>) =
        ids.joinToString(",", "[", "]") { id ->
            """{"id":$id,"globalID":${id * 10},"time":"2026-01-01T00:00:00Z",""" +
                """"type":"LocalChangeDetected","data":{"path":"a.org","folder":"abc-def",""" +
                """"action":"modified","type":"file"}}"""
        }

    private fun capturedRequests(block: suspend (KtorSyncthingClient) -> Unit): List<Url> {
        val urls = mutableListOf<Url>()
        val engine = MockEngine { request ->
            urls.add(request.url)
            respondOk(diskEventsJson(listOf(1)))
        }
        val client = KtorSyncthingClient(
            baseUrl = "http://127.0.0.1:8384",
            apiKey = "test-key",
            pollTimeoutSeconds = 60,
            engine = engine,
        )
        kotlinx.coroutines.runBlocking { block(client) }
        engine.close()
        return urls
    }

    @Test
    fun `events poll requests disk-event mask with since and timeout`() = runTest {
        val urls = capturedRequests { it.events(42, 60) }
        assertEquals(1, urls.size)
        val url = urls[0]
        assertEquals("/rest/events", url.encodedPath)
        assertEquals(
            "LocalChangeDetected,RemoteChangeDetected",
            url.parameters["events"],
        )
        assertEquals("42", url.parameters["since"])
        assertEquals("60", url.parameters["timeout"])
    }

    @Test
    fun `latestEventId seeds from the same disk-event mask`() = runTest {
        val urls = capturedRequests { it.latestEventId() }
        assertEquals(1, urls.size)
        val url = urls[0]
        assertEquals("/rest/events", url.encodedPath)
        // Regression for the silent-stall incident: without this filter the
        // cursor was seeded from the default subscription's counter, which
        // belongs to a different per-subscription ID space than the poll.
        assertEquals(
            "LocalChangeDetected,RemoteChangeDetected",
            url.parameters["events"],
        )
        assertEquals("0", url.parameters["since"])
        assertEquals("1", url.parameters["limit"])
    }

    @Test
    fun `events decodes id from disk event json`() = runTest {
        val engine = MockEngine { respondOk(diskEventsJson(listOf(7))) }
        val client = KtorSyncthingClient(
            baseUrl = "http://127.0.0.1:8384",
            apiKey = "test-key",
            pollTimeoutSeconds = 60,
            engine = engine,
        )
        kotlinx.coroutines.runBlocking {
            assertEquals(listOf(7L), client.events(0, 60).map { it.id })
        }
        engine.close()
    }

    @Test
    fun `client api key header is set`() = runTest {
        val headers = mutableListOf<String>()
        val engine = MockEngine { request ->
            headers.add(request.headers["X-API-Key"] ?: "")
            respondOk(diskEventsJson(listOf(1)))
        }
        val client = KtorSyncthingClient(
            baseUrl = "http://127.0.0.1:8384",
            apiKey = "sekrit",
            pollTimeoutSeconds = 60,
            engine = engine,
        )
        kotlinx.coroutines.runBlocking { client.events(0, 60) }
        engine.close()
        assertTrue(headers.all { it == "sekrit" })
    }
}