Initial commit
This commit is contained in:
weishaupt
@@ -0,0 +1,343 @@
|
||||
package irc
|
||||
|
||||
import data.*
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.SupervisorJob
|
||||
import kotlinx.coroutines.delay
|
||||
import kotlinx.coroutines.flow.MutableSharedFlow
|
||||
import kotlinx.coroutines.flow.MutableStateFlow
|
||||
import kotlinx.coroutines.flow.SharedFlow
|
||||
import kotlinx.coroutines.flow.StateFlow
|
||||
import kotlinx.coroutines.flow.update
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.sync.Mutex
|
||||
import kotlinx.coroutines.sync.withLock
|
||||
import java.io.File
|
||||
import java.util.concurrent.ConcurrentHashMap
|
||||
|
||||
// Per-buffer message history cap. Named so it's a one-line change to temporarily lower
|
||||
// for testing what happens once a busy channel's buffer actually hits the cap (auto-scroll
|
||||
// keying on the wrong thing is invisible in a quiet test channel that never reaches this).
|
||||
private const val MESSAGE_HISTORY_CAP = 500
|
||||
|
||||
// How long to wait after a message before actually writing that buffer's history to disk,
|
||||
// coalescing any further messages that arrive in the meantime into the same write. See
|
||||
// scheduleSave() - previously every single message triggered its own full save, which is
|
||||
// a lot of disk I/O to keep up with in a busy channel and was the likely direct cause of
|
||||
// reported slowness under real message volume.
|
||||
private const val SAVE_DEBOUNCE_MS = 2500L
|
||||
|
||||
/**
|
||||
* App-wide singleton. Owns every IrcConnection and every ChannelBuffer, so that all
|
||||
* windows read from the same source of truth via StateFlow.
|
||||
*/
|
||||
object ConnectionManager {
|
||||
private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
|
||||
private val connections = ConcurrentHashMap<String, IrcConnection>()
|
||||
private val ownNicks = ConcurrentHashMap<String, String>() // serverId -> our current nick
|
||||
private var historyRepo: MessageHistoryRepository? = null
|
||||
|
||||
private val _buffers = MutableStateFlow<Map<BufferKey, ChannelBuffer>>(emptyMap())
|
||||
val buffers: StateFlow<Map<BufferKey, ChannelBuffer>> = _buffers
|
||||
|
||||
private val _statuses = MutableStateFlow<Map<String, String>>(emptyMap())
|
||||
val statuses: StateFlow<Map<String, String>> = _statuses
|
||||
|
||||
private val _ownNicks = MutableStateFlow<Map<String, String>>(emptyMap())
|
||||
val ownNicksFlow: StateFlow<Map<String, String>> = _ownNicks
|
||||
|
||||
fun init(rootDir: File) {
|
||||
historyRepo = MessageHistoryRepository(rootDir)
|
||||
}
|
||||
|
||||
private fun setOwnNick(serverId: String, nick: String) {
|
||||
ownNicks[serverId] = nick
|
||||
_ownNicks.update { it + (serverId to nick) }
|
||||
}
|
||||
|
||||
// Fires once for every brand-new private-message buffer (a query with someone
|
||||
// who wasn't already an open window). Main.kt listens to this to auto-open a
|
||||
// window for it, same as clicking a channel manually.
|
||||
private val _newQueryBuffers = MutableSharedFlow<BufferKey>(extraBufferCapacity = 16)
|
||||
val newQueryBuffers: SharedFlow<BufferKey> = _newQueryBuffers
|
||||
|
||||
fun connect(config: ServerConfig) {
|
||||
// TEMP DEBUG (see task: on-connect identify command investigation) - confirms
|
||||
// whether the config this connect() call actually received has the on-connect
|
||||
// commands the caller thinks it saved.
|
||||
println("[connmgr-debug] connect(${config.id}) called - onConnectCommands=${config.onConnectCommands}")
|
||||
val existing = connections[config.id]
|
||||
if (existing != null) {
|
||||
// This used to just `return` here, making connect() a permanent no-op for the
|
||||
// rest of the process once a server had been connected once - editing that
|
||||
// server's config afterward (identify commands, autojoin, etc.) and re-saving
|
||||
// had zero effect until a full app restart, because the running IrcConnection's
|
||||
// `config` was captured once at construction and never replaced. Tearing down
|
||||
// the old connection and replacing it means a connect() call always actually
|
||||
// uses whatever was most recently saved.
|
||||
println("[connmgr-debug] connect(${config.id}) - already connected, replacing with a fresh connection using the latest config")
|
||||
existing.disconnect()
|
||||
}
|
||||
setOwnNick(config.id, config.nick)
|
||||
val conn = IrcConnection(
|
||||
config = config,
|
||||
scope = scope,
|
||||
onLine = { line -> handleLine(config.id, line) },
|
||||
onStatus = { msg -> _statuses.update { it + (config.id to msg) } }
|
||||
)
|
||||
connections[config.id] = conn
|
||||
conn.connect()
|
||||
}
|
||||
|
||||
fun disconnectAll() {
|
||||
connections.values.forEach { it.disconnect() }
|
||||
connections.clear()
|
||||
flushAllSaves()
|
||||
}
|
||||
|
||||
fun connectionFor(serverId: String): IrcConnection? = connections[serverId]
|
||||
|
||||
// IRC servers don't echo PRIVMSG back to its sender, so a sent message has to be
|
||||
// added to our own buffer locally - otherwise it would never show up in the window
|
||||
// that sent it.
|
||||
fun sendChannelMessage(serverId: String, target: String, text: String) {
|
||||
connectionFor(serverId)?.sendMessage(target, text)
|
||||
val myNick = ownNicks[serverId] ?: return
|
||||
addMessage(
|
||||
BufferKey(serverId, target),
|
||||
IrcMessage(System.currentTimeMillis(), MessageKind.CHAT, myNick, text)
|
||||
)
|
||||
}
|
||||
|
||||
// Same local-echo problem as sendChannelMessage, but for /me: the wire form is a
|
||||
// CTCP ACTION wrapper, while the buffer should just hold the plain action text
|
||||
// (ChannelScreen renders MessageKind.ACTION with its own "* nick ..." prefix).
|
||||
fun sendChannelAction(serverId: String, target: String, action: String) {
|
||||
connectionFor(serverId)?.sendMessage(target, "ACTION $action")
|
||||
val myNick = ownNicks[serverId] ?: return
|
||||
addMessage(
|
||||
BufferKey(serverId, target),
|
||||
IrcMessage(System.currentTimeMillis(), MessageKind.ACTION, myNick, action)
|
||||
)
|
||||
}
|
||||
|
||||
// Raw passthrough for slash commands that don't have dedicated handling.
|
||||
fun sendRaw(serverId: String, raw: String) {
|
||||
connectionFor(serverId)?.send(raw)
|
||||
}
|
||||
|
||||
// Local-only feedback (e.g. the /reload confirmation) that never touches the
|
||||
// wire - just appended to the buffer the same way any other SYSTEM line is.
|
||||
fun addSystemMessage(serverId: String, target: String, text: String) {
|
||||
addMessage(BufferKey(serverId, target), IrcMessage(System.currentTimeMillis(), MessageKind.SYSTEM, null, text))
|
||||
}
|
||||
|
||||
// One mutex per buffer key, guarding the "is this buffer new -> seed its history from
|
||||
// disk if so -> append the new message" sequence for that key end to end. Previously
|
||||
// that sequence was: a synchronous isNewBuffer read, then a separate async coroutine
|
||||
// that did the disk seed-load and the in-memory update - two steps with a real window
|
||||
// between them. In a busy channel, several messages for a brand-new buffer (e.g. right
|
||||
// after JOIN, when NAMES/TOPIC/a flood of chat lines all land within milliseconds) could
|
||||
// each see isNewBuffer=true before any of them had finished writing, each independently
|
||||
// re-reading the same stale on-disk snapshot and racing to write - concretely wasteful
|
||||
// (redundant disk reads) and, since MessageHistoryRepository.saveMessages() does a plain
|
||||
// unsynchronized file.writeText() with no ordering guarantee between concurrent callers,
|
||||
// capable of persisting a stale/incomplete snapshot last even when the in-memory state
|
||||
// was already fully correct. A per-key Mutex makes the whole check-seed-append sequence
|
||||
// for a given key strictly one-at-a-time, so only the genuinely first caller ever seeds
|
||||
// from disk and no two writers for the same key can be in that window simultaneously.
|
||||
private val bufferMutexes = ConcurrentHashMap<BufferKey, Mutex>()
|
||||
private fun mutexFor(key: BufferKey): Mutex = bufferMutexes.computeIfAbsent(key) { Mutex() }
|
||||
|
||||
// Pending debounced saves, keyed by buffer - see scheduleSave().
|
||||
private val pendingSaveJobs = ConcurrentHashMap<BufferKey, Job>()
|
||||
|
||||
private fun addMessage(key: BufferKey, msg: IrcMessage) {
|
||||
// Do the file I/O (seeding a brand-new buffer from disk, then persisting the
|
||||
// updated list) off the caller's thread - addMessage gets called both from the
|
||||
// IO-dispatched read loop and directly from UI-thread local-echo sends.
|
||||
scope.launch(Dispatchers.IO) {
|
||||
mutexFor(key).withLock {
|
||||
val isNewBuffer = !_buffers.value.containsKey(key)
|
||||
val seeded = if (isNewBuffer) (historyRepo?.loadMessages(key) ?: emptyList()) else emptyList()
|
||||
_buffers.update { map ->
|
||||
val existing = map[key] ?: ChannelBuffer(key, messages = seeded)
|
||||
val trimmed = (existing.messages + msg).takeLast(MESSAGE_HISTORY_CAP)
|
||||
map + (key to existing.copy(messages = trimmed))
|
||||
}
|
||||
if (isNewBuffer && !isChannelName(key.name)) {
|
||||
_newQueryBuffers.emit(key)
|
||||
}
|
||||
}
|
||||
}
|
||||
scheduleSave(key)
|
||||
}
|
||||
|
||||
// Coalesces rapid-fire saves for the same buffer into at most one write every
|
||||
// SAVE_DEBOUNCE_MS, instead of a full history save (up to 500 messages) on every single
|
||||
// incoming message - the direct cause of reported slowness in a busy channel. Reads
|
||||
// _buffers.value fresh at write time (not at schedule time), so whichever message
|
||||
// triggers the eventual write, it always persists the latest state, not a stale
|
||||
// snapshot from whenever it happened to be scheduled. computeIfAbsent is atomic, so
|
||||
// concurrent callers for the same key can only ever schedule one pending job.
|
||||
private fun scheduleSave(key: BufferKey) {
|
||||
pendingSaveJobs.computeIfAbsent(key) {
|
||||
scope.launch(Dispatchers.IO) {
|
||||
delay(SAVE_DEBOUNCE_MS)
|
||||
pendingSaveJobs.remove(key)
|
||||
val toSave = _buffers.value[key]?.messages ?: emptyList()
|
||||
historyRepo?.saveMessages(key, toSave)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Bypasses the debounce and writes every buffer's current state immediately - called on
|
||||
// shutdown so the last few seconds of a debounce window are never silently lost.
|
||||
private fun flushAllSaves() {
|
||||
pendingSaveJobs.values.forEach { it.cancel() }
|
||||
pendingSaveJobs.clear()
|
||||
_buffers.value.forEach { (key, buffer) -> historyRepo?.saveMessages(key, buffer.messages) }
|
||||
}
|
||||
|
||||
private fun updateUsers(key: BufferKey, transform: (List<ChannelUser>) -> List<ChannelUser>) {
|
||||
_buffers.update { map ->
|
||||
val existing = map[key] ?: ChannelBuffer(key)
|
||||
map + (key to existing.copy(users = transform(existing.users)))
|
||||
}
|
||||
}
|
||||
|
||||
private fun handleLine(serverId: String, line: ParsedLine) {
|
||||
val now = System.currentTimeMillis()
|
||||
val nick = nickFromPrefix(line.prefix) ?: line.prefix
|
||||
|
||||
when (line.command) {
|
||||
"PRIVMSG" -> {
|
||||
val target = line.params.getOrNull(0) ?: return
|
||||
val text = line.params.getOrNull(1) ?: return
|
||||
val isAction = text.startsWith("ACTION") && text.endsWith("")
|
||||
val body = if (isAction) text.removePrefix("ACTION ").removeSuffix("") else text
|
||||
// Channel messages are keyed by the channel name. Private messages arrive
|
||||
// with target == our own nick, so key those by the *sender* instead -
|
||||
// that's the conversation identity a query window should use.
|
||||
val bufferName = if (isChannelName(target)) target else (nick ?: target)
|
||||
addMessage(
|
||||
BufferKey(serverId, bufferName),
|
||||
IrcMessage(now, if (isAction) MessageKind.ACTION else MessageKind.CHAT, nick, body)
|
||||
)
|
||||
}
|
||||
"JOIN" -> {
|
||||
val channel = line.params.getOrNull(0) ?: return
|
||||
addMessage(BufferKey(serverId, channel), IrcMessage(now, MessageKind.JOIN, nick, "$nick joined $channel"))
|
||||
if (nick != null) updateUsers(BufferKey(serverId, channel)) { it + ChannelUser(nick) }
|
||||
}
|
||||
"PART" -> {
|
||||
val channel = line.params.getOrNull(0) ?: return
|
||||
addMessage(BufferKey(serverId, channel), IrcMessage(now, MessageKind.PART, nick, "$nick left $channel"))
|
||||
if (nick != null) updateUsers(BufferKey(serverId, channel)) { users -> users.filterNot { it.nick == nick } }
|
||||
}
|
||||
"KICK" -> {
|
||||
val channel = line.params.getOrNull(0) ?: return
|
||||
val kicked = line.params.getOrNull(1) ?: return
|
||||
val reason = line.params.getOrNull(2) ?: ""
|
||||
addMessage(
|
||||
BufferKey(serverId, channel),
|
||||
IrcMessage(now, MessageKind.KICK, nick, "$kicked was kicked by $nick ($reason)", target = kicked)
|
||||
)
|
||||
updateUsers(BufferKey(serverId, channel)) { users -> users.filterNot { it.nick == kicked } }
|
||||
}
|
||||
"MODE" -> {
|
||||
val channel = line.params.getOrNull(0) ?: return
|
||||
if (!channel.startsWith("#")) return
|
||||
val modeStr = line.params.getOrNull(1) ?: ""
|
||||
val modeTarget = line.params.getOrNull(2)
|
||||
addMessage(
|
||||
BufferKey(serverId, channel),
|
||||
IrcMessage(now, MessageKind.MODE, nick, "$nick sets mode $modeStr ${modeTarget ?: ""}".trim())
|
||||
)
|
||||
if (modeTarget != null && (modeStr == "+o" || modeStr == "-o" || modeStr == "+v" || modeStr == "-v")) {
|
||||
updateUsers(BufferKey(serverId, channel)) { users ->
|
||||
users.map { u ->
|
||||
if (u.nick != modeTarget) u
|
||||
else when (modeStr) {
|
||||
"+o" -> u.copy(isOp = true)
|
||||
"-o" -> u.copy(isOp = false)
|
||||
"+v" -> u.copy(isVoice = true)
|
||||
"-v" -> u.copy(isVoice = false)
|
||||
else -> u
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
"TOPIC" -> {
|
||||
val channel = line.params.getOrNull(0) ?: return
|
||||
val topic = line.params.getOrNull(1) ?: ""
|
||||
_buffers.update { map ->
|
||||
val key = BufferKey(serverId, channel)
|
||||
val existing = map[key] ?: ChannelBuffer(key)
|
||||
map + (key to existing.copy(topic = topic))
|
||||
}
|
||||
addMessage(BufferKey(serverId, channel), IrcMessage(now, MessageKind.TOPIC, nick, "$nick changed topic: $topic"))
|
||||
}
|
||||
"353" -> { // NAMES reply
|
||||
val channel = line.params.getOrNull(2) ?: return
|
||||
val names = line.params.getOrNull(3)?.split(" ")?.filter { it.isNotBlank() } ?: return
|
||||
val parsedUsers = names.map { raw ->
|
||||
when {
|
||||
raw.startsWith("@") -> ChannelUser(raw.removePrefix("@"), isOp = true)
|
||||
raw.startsWith("+") -> ChannelUser(raw.removePrefix("+"), isVoice = true)
|
||||
else -> ChannelUser(raw)
|
||||
}
|
||||
}
|
||||
updateUsers(BufferKey(serverId, channel)) { existing -> (existing + parsedUsers).distinctBy { it.nick } }
|
||||
}
|
||||
"NOTICE" -> {
|
||||
val target = line.params.getOrNull(0) ?: return
|
||||
val text = line.params.getOrNull(1) ?: return
|
||||
// Same fix PRIVMSG already had: a notice directed AT us arrives with
|
||||
// target == our own nick, not the sender. This mattered a lot in practice -
|
||||
// NickServ/ChanServ almost universally reply to commands (IDENTIFY, etc.)
|
||||
// via NOTICE rather than PRIVMSG, so without this fallback every identify
|
||||
// success/failure notice was being filed under a buffer keyed by our OWN
|
||||
// nick - which no window ever opens for - instead of under "NickServ",
|
||||
// making it look like the automatic identify produced no feedback at all,
|
||||
// whether or not it actually worked.
|
||||
val bufferName = if (isChannelName(target)) target else (nick ?: target)
|
||||
addMessage(BufferKey(serverId, bufferName), IrcMessage(now, MessageKind.NOTICE, nick, text))
|
||||
}
|
||||
"NICK" -> {
|
||||
val oldNick = nick ?: return
|
||||
val newNick = line.params.getOrNull(0) ?: return
|
||||
val affectedKeys = _buffers.value.filterKeys { it.serverId == serverId }
|
||||
.filterValues { buf -> buf.users.any { it.nick == oldNick } }
|
||||
.keys
|
||||
_buffers.update { map ->
|
||||
map.mapValues { (key, buf) ->
|
||||
if (key.serverId == serverId && buf.users.any { it.nick == oldNick }) {
|
||||
buf.copy(users = buf.users.map { u -> if (u.nick == oldNick) u.copy(nick = newNick) else u })
|
||||
} else buf
|
||||
}
|
||||
}
|
||||
affectedKeys.forEach { key ->
|
||||
addMessage(key, IrcMessage(now, MessageKind.SYSTEM, null, "$oldNick is now known as $newNick"))
|
||||
}
|
||||
if (ownNicks[serverId] == oldNick) {
|
||||
setOwnNick(serverId, newNick)
|
||||
}
|
||||
}
|
||||
"QUIT" -> {
|
||||
val reason = line.params.getOrNull(0) ?: ""
|
||||
// Remove nick from every channel buffer on this server it appears in.
|
||||
_buffers.update { map ->
|
||||
map.mapValues { (key, buf) ->
|
||||
if (key.serverId == serverId && nick != null && buf.users.any { it.nick == nick }) {
|
||||
buf.copy(users = buf.users.filterNot { it.nick == nick })
|
||||
} else buf
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user