// // ChatStore.swift // AT-IRC // // App state: OAuth sign-in, atirc token minting, live subscription, // per-channel cursor scrollback, and posting. // import Foundation import Observation import OSLog @Observable @MainActor final class ChatStore { enum Phase: Equatable { case signedOut case signingIn case connecting case connected case failed(String) } /// Cached messages and cursor for one channel. struct ChannelState { var messages: [AtircMessage] = [] var seenIds: Set = [] /// Highest sequence seen. Passed as the `cursor` when (re)subscribing /// so the server replays only messages we have not seen yet. var lastSeq: Int64 = 0 var historyLoading = false var historyLoaded = false } struct ChannelInfo: Identifiable, Hashable { var name: String var lastSeq: Int64 var id: String { name } } var phase: Phase = .signedOut var account: AtircAccount? var composerText: String = "" var statusText: String = "" var errorText: String? /// Owner questions, newest first, and the owner's current availability. /// `ownerMode` is `available` or `autonomous` (agents decide on their own). var asks: [AskQuestion] = [] var ownerMode: String = "available" var asksError: String? /// Channels seen in the server's retained history; own channel first, then /// most recently active. var channels: [ChannelInfo] = [] var selectedChannel: String = "" private var channelStates: [String: ChannelState] = [:] /// Connected identities per channel, from the last presence snapshot plus /// join/leave frames. Dropped whenever a new connection starts. private var channelMembers: [String: [AtircMember]] = [:] /// Inline join/leave notices per channel, capped like a scrollback window. private var channelNotices: [String: [ChatNotice]] = [:] /// Presence events buffered for one coalesced state update. private struct PendingPresence { let channel: String let member: AtircMember let joining: Bool let notice: ChatNotice } @ObservationIgnored private var pendingPresence: [PendingPresence] = [] @ObservationIgnored private var presenceFlushTask: Task? /// Newest messages per channel, captured during discovery. Used to paint a /// channel at the bottom before its full scrollback arrives. @ObservationIgnored private var recentMessages: [String: [AtircMessage]] = [:] @ObservationIgnored private var askPollTask: Task? @ObservationIgnored private var lastAskRefresh: Date? /// Ask ids already announced, so relaunches do not repeat notifications. @ObservationIgnored private var notifiedAskIDs: Set = Set(UserDefaults.standard.stringArray(forKey: "atirc.notifiedAskIDs.v1") ?? []) private let client = AtircClient() private let log = Logger(subsystem: "com.fedfork.irc.AT-IRC", category: "stream") private let presenceLog = Logger(subsystem: "com.fedfork.irc.AT-IRC", category: "presence") /// The first unexpected presence shape is logged once for diagnosis. @ObservationIgnored private var didLogPresenceShape = false private let oauth = OAuthClient() private var pending: OAuthClient.PendingAuthorization? private var socket: AtircWebSocket? @ObservationIgnored private var reconnectTask: Task? @ObservationIgnored private var reconnectAttempt = 0 @ObservationIgnored private var authRecoveryAttempt = 0 @ObservationIgnored private var isBackgrounded = false private let accountKey = "atirc.account.v1" private static let pageSize = 200 /// How many recent messages are shown the moment a channel is opened. private static let recentLimit = 10 /// Inline presence notices kept per channel. private static let noticeLimit = 200 /// The server retains 5000 messages by default; 25 pages covers all of it. private static let maxHistoryPages = 25 private var lastChannelScan: Date? private var isScanningChannels = false var channel: String { selectedChannel } var isSignedIn: Bool { account != nil } var messages: [AtircMessage] { channelStates[selectedChannel]?.messages ?? [] } /// Identities currently connected to the selected channel. var members: [AtircMember] { channelMembers[selectedChannel] ?? [] } /// Inline join/leave notices for the selected channel. var notices: [ChatNotice] { channelNotices[selectedChannel] ?? [] } /// Cached, already-ordered transcript per channel. Live rows are inserted /// incrementally and history merges rebuild once; view bodies never sort. private var timelines: [String: [ChatRow]] = [:] /// Messages and presence notices interleaved by time. var timeline: [ChatRow] { timelines[selectedChannel] ?? [] } /// Whether the selected channel's scrollback is still being fetched. var isLoadingHistory: Bool { channelStates[selectedChannel]?.historyLoading ?? false } static func orderedTimeline(_ messages: [AtircMessage], notices: [ChatNotice]) -> [ChatRow] { let rows = messages.map(ChatRow.message) + notices.map(ChatRow.notice) // Parse each timestamp once; the comparator runs O(n log n) times. let decorated = rows.enumerated().map { (offset: $0.offset, row: $0.element, date: $0.element.date, seq: $0.element.seq) } return decorated.sorted { lhs, rhs in if let left = lhs.date, let right = rhs.date, left != right { return left < right } if let left = lhs.seq, let right = rhs.seq, left != right { return left < right } return lhs.offset < rhs.offset }.map(\.row) } var pendingAsks: [AskQuestion] { asks.filter { $0.isPending } } var pendingAskCount: Int { pendingAsks.count } // MARK: - Lifecycle /// Load a persisted account, if any. func restore() { guard account == nil else { return } guard let data = Keychain.get(accountKey), let saved = try? JSONDecoder().decode(AtircAccount.self, from: data) else { return } account = saved selectedChannel = saved.channel ensureChannel(saved.channel) phase = .connecting } /// Prime channel discovery (which also caches each channel's newest /// messages), then open the selected channel. func start() async { guard let account else { return } applyDebugChannelOverride() if selectedChannel.isEmpty { selectedChannel = account.channel } ensureChannel(selectedChannel) log.notice("start channel=\(self.selectedChannel, privacy: .public) account=\(account.mint.subhandle, privacy: .public)") // Show the channel immediately: subscribe and load history first, then // run the (possibly slow) global channel discovery in the background. if socket == nil { activateSelectedChannel() } startAskPolling() Task { await refreshChannels(force: true) } } // MARK: - Sign-in /// Run the full OAuth flow. `present` is called with the authorization URL /// to show in a browser; this method returns once the account is ready. func signIn(handle: String, present: (URL) -> Void) async throws { phase = .signingIn errorText = nil do { let (url, pending) = try await oauth.beginSignIn(handle: handle) self.pending = pending present(url) let query = try await pending.loopback.waitForCallback() try await finishSignIn(query: query) } catch { pending = nil phase = .signedOut errorText = error.localizedDescription throw error } } private func finishSignIn(query: [String: String]) async throws { guard let pending else { throw OAuthError(message: "No pending sign-in") } self.pending = nil let oauthSession = try await oauth.completeSignIn(pending: pending, query: query) guard let handle = oauthSession.handle, !handle.isEmpty else { throw OAuthError(message: "Could not determine your handle") } let (mint, updatedSession) = try await client.mint(oauthSession: oauthSession, subhandle: handle) var session = updatedSession session.handle = mint.handle let channel = "#" + mint.handle.replacingOccurrences(of: ".", with: "_") let account = AtircAccount(oauth: session, mint: mint, channel: channel) self.account = account selectedChannel = channel ensureChannel(channel) applyDebugChannelOverride() persist() phase = .connecting activateSelectedChannel() Task { await refreshChannels(force: true) } startAskPolling() } func signOut() { stopAskPolling() reconnectTask?.cancel() reconnectTask = nil reconnectAttempt = 0 socket?.disconnect() socket = nil Keychain.delete(accountKey) account = nil channelStates = [:] channelMembers = [:] channelNotices = [:] timelines = [:] presenceFlushTask?.cancel() presenceFlushTask = nil pendingPresence.removeAll(keepingCapacity: true) didLogPresenceShape = false recentMessages = [:] channels = [] selectedChannel = "" statusText = "" errorText = nil lastChannelScan = nil asks = [] ownerMode = "available" asksError = nil lastAskRefresh = nil phase = .signedOut } // MARK: - Channel switching func openChannel(_ name: String) { guard name != selectedChannel else { return } socket?.disconnect() socket = nil selectedChannel = name ensureChannel(name) errorText = nil statusText = "" activateSelectedChannel() } /// Show the newest cached messages immediately, then join for live updates /// and load the full scrollback in the background. private func activateSelectedChannel() { let name = selectedChannel guard !name.isEmpty else { return } seedRecentMessages(for: name) rebuildTimeline(for: name) connect() startHistoryLoad(for: name) } /// Paint the channel with its newest discovered messages so it opens at /// the bottom instead of scrolling through the whole history. private func seedRecentMessages(for name: String) { guard let recent = recentMessages[name], !recent.isEmpty else { return } guard channelStates[name]?.messages.isEmpty ?? true else { return } mergeMessages(recent, into: name) } /// Scripted UI tests can open a specific channel: ATIRC_OPEN_CHANNEL. private func applyDebugChannelOverride() { #if DEBUG if let override = ProcessInfo.processInfo.environment["ATIRC_OPEN_CHANNEL"], !override.isEmpty { selectedChannel = override ensureChannel(override) } #endif } private func ensureChannel(_ name: String) { guard !name.isEmpty else { return } if channelStates[name] == nil { channelStates[name] = ChannelState() } if !channels.contains(where: { $0.name == name }) { channels.append(ChannelInfo(name: name, lastSeq: 0)) } } // MARK: - Subscription /// Reconnect with exponential backoff (1s doubling to 30s). A successful /// connection resets the delay. Server exec-upgrades and suspended /// sockets both surface here. private func scheduleReconnect() { guard account != nil, !selectedChannel.isEmpty, !isBackgrounded else { return } reconnectTask?.cancel() reconnectAttempt += 1 let delay = min(30.0, pow(2.0, Double(reconnectAttempt - 1))) log.notice("reconnect scheduled attempt=\(self.reconnectAttempt, privacy: .public) delay=\(delay, privacy: .public)s") statusText = String(format: "reconnecting in %.0fs…", delay) reconnectTask = Task { @MainActor [weak self] in try? await Task.sleep(for: .seconds(delay)) guard let self, !Task.isCancelled else { return } self.reconnectTask = nil self.connect() } } /// DEBUG: force a stream failure so scripted tests can observe the backoff. func simulateStreamDropForTest() { #if DEBUG guard ProcessInfo.processInfo.environment["ATIRC_TEST_DROP_STREAM"] == "1" else { return } socket?.disconnect() statusText = "error: simulated drop" phase = .failed(statusText) scheduleReconnect() #endif } /// Foreground/background transitions from the view. func setBackgrounded(_ background: Bool) { isBackgrounded = background if background { reconnectTask?.cancel() reconnectTask = nil } } /// Re-establish the live subscription after returning from the background. /// iOS tears down the URLSessionWebSocketTask while the app is suspended; /// the socket object survives but is dead, so reconnect from the last seen /// sequence and let the server replay what was missed. func resumeStream() { guard account != nil, !selectedChannel.isEmpty else { return } // Do not tear down a connection that is still being established; the // foreground event can race the initial subscribe. guard phase != .connecting else { return } log.notice("resumeStream channel=\(self.selectedChannel, privacy: .public)") reconnectTask?.cancel() reconnectTask = nil reconnectAttempt = 0 socket?.disconnect() socket = nil connect() } private func connect() { guard let account, !selectedChannel.isEmpty else { return } // Presence is snapshotted per connection. Drop stale nicks now and // wait for the server's next presence_snapshot. if (channelMembers[selectedChannel]?.isEmpty == false) { channelMembers[selectedChannel] = [] } presenceFlushTask?.cancel() presenceFlushTask = nil pendingPresence.removeAll(keepingCapacity: true) reconnectTask?.cancel() reconnectTask = nil socket?.disconnect() statusText = "connecting…" phase = .connecting ensureChannel(selectedChannel) let state = channelStates[selectedChannel] ?? ChannelState() log.notice("connect channel=\(self.selectedChannel, privacy: .public) cursor=\(state.lastSeq, privacy: .public)") let ws = AtircWebSocket() ws.onStatus = { [weak self] status in guard let self else { return } self.log.notice("stream status: \(status, privacy: .public)") self.statusText = status if status == "connected" { self.phase = .connected self.reconnectAttempt = 0 self.authRecoveryAttempt = 0 } else if status.hasPrefix("failed") || status.hasPrefix("error") || status.hasPrefix("closed") { if status.contains("403") || status.localizedCaseInsensitiveContains("forbidden") { // Not recoverable by retrying: the token cannot join. self.phase = .failed( "can't join \(self.selectedChannel) — you can only join channels you can write to" ) } else if status.contains("401") || status.localizedCaseInsensitiveContains("unauthorized") || status.contains("-1011") || status.localizedCaseInsensitiveContains("bad server response") { // The minted token can be invalidated by server restarts // (wave-3 iat cutoff) or revocation. A rejected WebSocket // upgrade surfaces as NSURLErrorBadServerResponse (-1011), // not an HTTP 401, so both mean re-mint once. if self.authRecoveryAttempt < 2 { self.authRecoveryAttempt += 1 Task { await self.reauth() } } else { self.phase = .failed(status) self.scheduleReconnect() } } else { self.phase = .failed(status) self.scheduleReconnect() } } } ws.onFrame = { [weak self] frame in self?.handle(frame) } // With no known messages, subscribe to new messages only; the initial // recent window comes from discovery and the scrollback from `list`. let cursor = state.messages.isEmpty ? -1 : state.lastSeq socket = ws ws.connect( url: subscribeURL(account: account, channel: selectedChannel, cursor: cursor), token: account.mint.token ) } private func subscribeURL(account: AtircAccount, channel: String, cursor: Int64) -> URL { let wsBase = account.atircBaseURL .replacingOccurrences(of: "https://", with: "wss://") .replacingOccurrences(of: "http://", with: "ws://") var components = URLComponents(string: wsBase)! components.path = "/xrpc/chat.atirc.subscribe" components.queryItems = [ URLQueryItem(name: "channel", value: channel), URLQueryItem(name: "cursor", value: String(cursor)), ] return components.url! } private func handle(_ frame: AtircFrame) { if let error = frame.error { switch error { case "ConsumerTooSlow": statusText = "subscriber lagged; reconnecting…" connect() case "TokenRevoked": statusText = "token revoked; re-minting…" Task { await reauth() } default: statusText = "stream error: \(error)" } return } if frame.type == "#presence" { handlePresence(frame.body) return } guard frame.type == "#message" else { return } guard let message = Self.decodeMessage(frame.body) else { return } mergeMessages([message], into: message.channel ?? selectedChannel) phase = .connected } private static func decodeMessage(_ body: [String: Any]) -> AtircMessage? { guard let data = try? JSONSerialization.data(withJSONObject: body) else { return nil } return try? JSONDecoder().decode(AtircMessage.self, from: data) } // MARK: - Presence #if DEBUG /// Tests drive presence frames without a socket. func handlePresenceForTest(_ body: [String: Any]) { handlePresence(body) } #endif /// Presence arrives on the subscribe socket: a snapshot right after /// connect, then join/leave frames for the selected channel. private func handlePresence(_ body: [String: Any]) { guard let event = (body["event"] as? String)?.lowercased() else { logUnexpectedPresence(body, event: nil) return } switch event { case "presence_snapshot": applyPresenceSnapshot(body) case "join": applyPresenceEvent(body, joining: true) case "leave", "part": applyPresenceEvent(body, joining: false) default: logUnexpectedPresence(body, event: event) } } private func applyPresenceSnapshot(_ body: [String: Any]) { // Only subscribed channels are forwarded, but honor the explicit // channel list when it is present. if let channels = body["channels"] as? [String], !channels.contains(selectedChannel) { return } guard let raw = body["identities"] as? [Any] else { logUnexpectedPresence(body, event: "presence_snapshot") return } var seen = Set() var members: [AtircMember] = [] var recognizedAll = true for item in raw { guard let member = Self.member(from: item) else { recognizedAll = false continue } guard seen.insert(member.id).inserted else { continue } members.append(member) } if !recognizedAll { logUnexpectedPresence(body, event: "presence_snapshot") } let sorted = Self.sorted(members) if sorted != (channelMembers[selectedChannel] ?? []) { channelMembers[selectedChannel] = sorted } } private func applyPresenceEvent(_ body: [String: Any], joining: Bool) { if let channel = body["channel"] as? String, channel != selectedChannel { return } let nick = (body["nick"] as? String)?.trimmingCharacters(in: .whitespacesAndNewlines) let identity = (body["identity"] as? String)?.trimmingCharacters(in: .whitespacesAndNewlines) let rawIdentity = identity?.isEmpty == false ? identity : nick guard let member = rawIdentity.flatMap({ AtircMember.from(presenceIdentity: $0) }) else { logUnexpectedPresence(body, event: joining ? "join" : "leave") return } let displayNick = nick.flatMap { $0.isEmpty ? nil : $0 } ?? member.nick let display = AtircMember(id: member.id, nick: displayNick, identity: member.identity) let notice = ChatNotice( channel: selectedChannel, nick: displayNick, identity: display.identity, kind: joining ? .join : .leave ) // This channel churns joins/leaves (sub reconnect loops, synthetic pub // posters). Coalesce into one state update per ~250 ms so view bodies // and the toolbar badge do not re-render per frame. pendingPresence.append(PendingPresence( channel: selectedChannel, member: display, joining: joining, notice: notice )) schedulePresenceFlush() } private func schedulePresenceFlush() { guard presenceFlushTask == nil else { return } presenceFlushTask = Task { @MainActor [weak self] in try? await Task.sleep(for: .milliseconds(250)) guard let self, !Task.isCancelled else { return } self.flushPresence() } } /// Apply buffered presence events in one pass. Members publish only when /// the resulting list actually changed. private func flushPresence() { presenceFlushTask = nil guard !pendingPresence.isEmpty else { return } let updates = pendingPresence pendingPresence.removeAll(keepingCapacity: true) var members = channelMembers[selectedChannel] ?? [] var membersChanged = false var notices = channelNotices[selectedChannel] ?? [] var newRows: [ChatRow] = [] for update in updates { if update.joining { if let index = members.firstIndex(where: { $0.id == update.member.id }) { if members[index] != update.member { members[index] = update.member membersChanged = true } } else { members.append(update.member) membersChanged = true } } else { let before = members.count members.removeAll { $0.id == update.member.id } if members.count != before { membersChanged = true } } if update.channel == selectedChannel { notices.append(update.notice) newRows.append(.notice(update.notice)) } else { // The channel changed during the coalescing window; keep the // notice for that channel but leave its member list stale // (the next snapshot on subscribe refills it). var other = channelNotices[update.channel] ?? [] other.append(update.notice) if other.count > Self.noticeLimit { other.removeFirst(other.count - Self.noticeLimit) } channelNotices[update.channel] = other } } if membersChanged { let sorted = Self.sorted(members) if sorted != (channelMembers[selectedChannel] ?? []) { channelMembers[selectedChannel] = sorted } } if !newRows.isEmpty { if notices.count > Self.noticeLimit { notices.removeFirst(notices.count - Self.noticeLimit) } channelNotices[selectedChannel] = notices insertRows(newRows, into: selectedChannel) } } /// A snapshot identity is normally a `nick@host` string; accept objects /// too, in case the server grows fields. private static func member(from item: Any) -> AtircMember? { if let identity = item as? String { return AtircMember.from(presenceIdentity: identity) } guard let object = item as? [String: Any] else { return nil } func string(_ key: String) -> String? { guard let raw = object[key] as? String else { return nil } let value = raw.trimmingCharacters(in: .whitespacesAndNewlines) return value.isEmpty ? nil : value } let identity = string("identity") ?? string("handle") ?? string("did") ?? string("subhandle") guard let nick = string("nick") ?? identity else { return nil } let id = identity ?? nick return AtircMember(id: id, nick: nick, identity: identity == nick ? nil : identity) } private static func sorted(_ members: [AtircMember]) -> [AtircMember] { members.sorted { lhs, rhs in let order = lhs.nick.localizedCaseInsensitiveCompare(rhs.nick) if order != .orderedSame { return order == .orderedAscending } return lhs.id < rhs.id } } // MARK: - Transcript cache /// Rebuild one channel's ordered transcript (history merges and channel /// switches only). Live rows use `insertRows`. private func rebuildTimeline(for name: String) { guard !name.isEmpty else { return } let messages = channelStates[name]?.messages ?? [] let notices = channelNotices[name] ?? [] timelines[name] = Self.orderedTimeline(messages, notices: notices) } /// Insert live rows into the cached transcript, keeping (date, seq) order. private func insertRows(_ rows: [ChatRow], into name: String) { guard name == selectedChannel else { return } guard var list = timelines[name] else { rebuildTimeline(for: name) return } for row in rows { insertRow(row, into: &list) } timelines[name] = list } private func insertRow(_ row: ChatRow, into list: inout [ChatRow]) { // The common case is a live row newer than everything cached. if let last = list.last, Self.rowOrdered(row, before: last) == false { list.append(row) return } var low = 0 var high = list.count while low < high { let mid = (low + high) / 2 if Self.rowOrdered(row, before: list[mid]) { high = mid } else { low = mid + 1 } } list.insert(row, at: low) } /// Strict "a before b" using timestamp, then sequence. Equal rows answer /// false so inserts land after existing equals. private static func rowOrdered(_ a: ChatRow, before b: ChatRow) -> Bool { if let left = a.date, let right = b.date, left != right { return left < right } if let left = a.seq, let right = b.seq, left != right { return left < right } return false } /// One diagnostic line with the body's key names (never values) the first /// time the server sends a presence shape this client does not know. private func logUnexpectedPresence(_ body: [String: Any], event: String?) { guard !didLogPresenceShape else { return } didLogPresenceShape = true let keys = body.keys.sorted().joined(separator: ",") presenceLog.notice("unrecognized presence event=\(event ?? "none", privacy: .public) keys=\(keys, privacy: .public)") } // MARK: - Channel discovery /// Rebuild the channel list from the server's retained history. The list /// endpoint is unauthenticated and, without a channel filter, returns /// messages from every retained channel. func refreshChannels(force: Bool = false) async { guard let account, !isScanningChannels else { return } if !force, let last = lastChannelScan, Date().timeIntervalSince(last) < 20 { return } isScanningChannels = true defer { isScanningChannels = false } var latest: [String: Int64] = [:] var recent: [String: [AtircMessage]] = [:] var cursor: Int64 = 0 var pages = 0 do { while pages < Self.maxHistoryPages { let response = try await client.list( channel: nil, cursor: cursor, limit: Self.pageSize, baseURL: account.atircBaseURL ) for message in response.messages { guard let name = message.channel else { continue } latest[name] = max(latest[name] ?? 0, message.seq ?? 0) var buffer = recent[name] ?? [] buffer.append(message) if buffer.count > Self.recentLimit { buffer.removeFirst(buffer.count - Self.recentLimit) } recent[name] = buffer } guard !response.messages.isEmpty, response.cursor > cursor else { break } cursor = response.cursor pages += 1 if response.messages.count < Self.pageSize { break } } lastChannelScan = Date() for (name, messages) in recent { recentMessages[name] = messages } } catch { // Discovery is best-effort; keep the channels we already know. } var merged: [String: Int64] = [:] for info in channels { merged[info.name] = info.lastSeq } for (name, seq) in latest { merged[name] = max(merged[name] ?? 0, seq) } merged[account.channel] = max(merged[account.channel] ?? 0, 0) if !selectedChannel.isEmpty { merged[selectedChannel] = max(merged[selectedChannel] ?? 0, 0) } channels = merged.map { ChannelInfo(name: $0.key, lastSeq: $0.value) } sortChannels() log.notice("refresh channels=\(self.channels.count, privacy: .public) recent=\(recent.count, privacy: .public)") } private func noteChannel(_ name: String, lastSeq: Int64) { if let index = channels.firstIndex(where: { $0.name == name }) { if lastSeq > channels[index].lastSeq { channels[index].lastSeq = lastSeq } } else { channels.append(ChannelInfo(name: name, lastSeq: lastSeq)) sortChannels() } } private func sortChannels() { let own = account?.channel channels.sort { a, b in if a.name == own { return b.name != own } if b.name == own { return false } if a.lastSeq != b.lastSeq { return a.lastSeq > b.lastSeq } return a.name < b.name } } // MARK: - Scrollback /// Merge a batch of messages into a channel, keeping order and dropping /// duplicates (live frames, subscribe catch-up, and list pages overlap). private func mergeMessages(_ incoming: [AtircMessage], into name: String) { guard !name.isEmpty, !incoming.isEmpty else { return } var state = channelStates[name] ?? ChannelState() let wasEmpty = state.messages.isEmpty let previousLastSeq = state.messages.last?.seq ?? 0 var added = false var addedRows: [ChatRow] = [] for message in incoming { guard !state.seenIds.contains(message.id) else { continue } state.seenIds.insert(message.id) state.messages.append(message) addedRows.append(.message(message)) if let seq = message.seq, seq > state.lastSeq { state.lastSeq = seq } added = true } guard added else { return } // Batches arrive in seq order; only an older batch needs re-sorting. if !wasEmpty, (incoming.first?.seq ?? 0) < previousLastSeq { state.messages.sort { ($0.seq ?? 0) < ($1.seq ?? 0) } } channelStates[name] = state if name == selectedChannel { if wasEmpty || (incoming.first?.seq ?? 0) >= previousLastSeq { insertRows(addedRows, into: name) } else { rebuildTimeline(for: name) } } else { timelines.removeValue(forKey: name) } noteChannel(name, lastSeq: state.lastSeq) log.notice("merge channel=\(name, privacy: .public) added=\(incoming.count, privacy: .public) total=\(state.messages.count, privacy: .public)") } private func startHistoryLoad(for name: String) { log.notice("history start channel=\(name, privacy: .public)") var state = channelStates[name] ?? ChannelState() guard !state.historyLoaded, !state.historyLoading else { log.notice("history skip channel=\(name, privacy: .public) loaded=\(state.historyLoaded, privacy: .public) loading=\(state.historyLoading, privacy: .public)") return } state.historyLoading = true channelStates[name] = state Task { await loadHistory(name: name) } } /// Page the channel's retained history with chat.atirc.list's cursor. The /// newest messages are already on screen, so the full set is merged in a /// single update and the view stays anchored at the bottom. private func loadHistory(name: String) async { guard let account else { return } var collected: [AtircMessage] = [] var cursor: Int64 = 0 var pages = 0 var complete = false do { while pages < Self.maxHistoryPages { let response = try await client.list( channel: name, cursor: cursor, limit: Self.pageSize, baseURL: account.atircBaseURL ) collected.append(contentsOf: response.messages) log.notice("history page channel=\(name, privacy: .public) cursorIn=\(cursor, privacy: .public) count=\(response.messages.count, privacy: .public) cursorOut=\(response.cursor, privacy: .public)") guard !response.messages.isEmpty else { complete = true break } cursor = response.cursor pages += 1 if response.messages.count < Self.pageSize { complete = true break } } if pages >= Self.maxHistoryPages { complete = true } } catch { log.error("history failed channel=\(name, privacy: .public) error=\(error.localizedDescription, privacy: .public)") errorText = "history unavailable: \(error.localizedDescription)" } log.notice("history done channel=\(name, privacy: .public) pages=\(pages, privacy: .public) collected=\(collected.count, privacy: .public) complete=\(complete, privacy: .public)") mergeMessages(collected, into: name) var state = channelStates[name] ?? ChannelState() state.historyLoading = false state.historyLoaded = complete channelStates[name] = state } // MARK: - Posting func send() async { let text = composerText.trimmingCharacters(in: .whitespacesAndNewlines) guard !text.isEmpty, let account, !selectedChannel.isEmpty else { return } let target = selectedChannel composerText = "" do { let message = try await client.post( text: text, channel: target, token: account.mint.token, baseURL: account.atircBaseURL ) mergeMessages([message], into: message.channel ?? target) } catch let error as AtircHTTPError where error.isUnauthorized { composerText = text await reauth() await send() } catch is DecodingError { // The server accepted the message (HTTP 2xx); only its echoed // body was an unexpected shape. Keep the composer clear and let // the live subscription deliver the message. } catch { composerText = text errorText = error.localizedDescription } } // MARK: - Owner questions /// Fetch questions and the owner mode. Polled at sign-in, on foreground, /// and every 60s while the app runs; the owner gets one local notification /// per newly seen pending question. func refreshAsks(force: Bool = false) async { guard account != nil else { return } if !force, let last = lastAskRefresh, Date().timeIntervalSince(last) < 10 { return } do { try await loadAsks() } catch let error as AtircHTTPError where error.isUnauthorized { // The minted token is missing the new endpoint scopes until the // server deploy lands, and server restarts invalidate tokens. // Re-mint once and retry. await reauth() do { try await loadAsks() } catch { asksError = error.localizedDescription } } catch { asksError = error.localizedDescription } } private func loadAsks() async throws { guard let account else { return } let response = try await client.questions(token: account.mint.token, baseURL: account.atircBaseURL) ownerMode = response.mode asks = response.questions asksError = nil lastAskRefresh = Date() announceNewAsks() } /// Notify once per never-before-seen pending question id, only while the /// owner is available: away means agents should decide on their own. private func announceNewAsks() { guard ownerMode != "autonomous" else { return } let freshIDs = Self.newPendingAskIDs(in: asks, notified: notifiedAskIDs) guard !freshIDs.isEmpty else { return } notifiedAskIDs.formUnion(asks.filter { $0.isPending }.map(\.id)) persistNotifiedAskIDs() for ask in asks.filter({ freshIDs.contains($0.id) }).prefix(5) { AskNotifier.shared.announce(ask) } } static func newPendingAskIDs(in questions: [AskQuestion], notified: Set) -> [String] { questions.filter { $0.isPending }.map(\.id).filter { !notified.contains($0) } } private func persistNotifiedAskIDs() { // ponytail: the last 200 ids is plenty; no need for a store. UserDefaults.standard.set(Array(notifiedAskIDs.suffix(200)), forKey: "atirc.notifiedAskIDs.v1") } func answer(_ ask: AskQuestion, answer: String, choice: Int?) async { let text = answer.trimmingCharacters(in: .whitespacesAndNewlines) guard !text.isEmpty || choice != nil, let account else { return } do { let updated = try await client.answer( questionID: ask.id, answer: text, choice: choice, token: account.mint.token, baseURL: account.atircBaseURL ) replaceAsk(updated) AskNotifier.shared.clear(ask.id) } catch let error as AtircHTTPError where error.isUnauthorized { await reauth() guard let account = self.account, let updated = try? await client.answer( questionID: ask.id, answer: text, choice: choice, token: account.mint.token, baseURL: account.atircBaseURL ) else { asksError = "Could not answer this question" return } replaceAsk(updated) AskNotifier.shared.clear(ask.id) } catch { asksError = error.localizedDescription } } private func replaceAsk(_ updated: AskQuestion) { if let index = asks.firstIndex(where: { $0.id == updated.id }) { asks[index] = updated } else { asks.insert(updated, at: 0) } } /// Toggle availability. `available` means questions wait for the owner; /// `autonomous` means agents research, decide, and record an ADR. func setOwnerMode(_ mode: String) async { guard let account else { return } let previous = ownerMode ownerMode = mode do { try await client.setOwnerMode(mode, token: account.mint.token, baseURL: account.atircBaseURL) announceNewAsks() } catch let error as AtircHTTPError where error.isUnauthorized { await reauth() if let account = self.account, (try? await client.setOwnerMode(mode, token: account.mint.token, baseURL: account.atircBaseURL)) != nil { announceNewAsks() return } ownerMode = previous } catch { ownerMode = previous asksError = error.localizedDescription } } func startAskPolling() { AskNotifier.shared.prepare() guard askPollTask == nil else { return } askPollTask = Task { [weak self] in while !Task.isCancelled { await self?.refreshAsks(force: true) try? await Task.sleep(for: .seconds(60)) } } } func stopAskPolling() { askPollTask?.cancel() askPollTask = nil } // MARK: - Token recovery /// Re-mint the atirc token. Tokens are invalidated when the atirc server /// restarts, so this is the normal recovery path after a 401. func reauth() async { guard var account else { return } do { let (mint, oauthSession) = try await client.mint( oauthSession: account.oauth, subhandle: account.mint.subhandle ) account.mint = mint account.oauth = oauthSession account.channel = "#" + mint.handle.replacingOccurrences(of: ".", with: "_") self.account = account persist() connect() } catch { errorText = error.localizedDescription phase = .failed(error.localizedDescription) } } private func persist() { guard let account, let data = try? JSONEncoder().encode(account) else { return } Keychain.set(data, for: accountKey) } }