diff --git a/Bugger.xcodeproj/project.pbxproj b/Bugger.xcodeproj/project.pbxproj index 7012ac0..29c6c8b 100644 --- a/Bugger.xcodeproj/project.pbxproj +++ b/Bugger.xcodeproj/project.pbxproj @@ -32,6 +32,7 @@ CB529549DB1DBB5F02884300 /* FeishuError.swift in Sources */ = {isa = PBXBuildFile; fileRef = 2A83E76FBB83849E4B6704D3 /* FeishuError.swift */; }; D2855F82E68FADA8BEA43086 /* AppStateService.swift in Sources */ = {isa = PBXBuildFile; fileRef = F1DD178095EFCE7BC121C1C4 /* AppStateService.swift */; }; 7E1F2A3B4C5D6E7F8091A2B3 /* BitableEventService.swift in Sources */ = {isa = PBXBuildFile; fileRef = 8F2E3D4C5B6A7988061A2B3C /* BitableEventService.swift */; }; + A1B2C3D4E5F60718293A4B5C /* CalibrationService.swift in Sources */ = {isa = PBXBuildFile; fileRef = 1A2B3C4D5E6F70819293A4B5 /* CalibrationService.swift */; }; DDB9EB345FEBB95DD9976364 /* BugChange.swift in Sources */ = {isa = PBXBuildFile; fileRef = 0B24E1E1D32F7B774D12DD85 /* BugChange.swift */; }; E4CBAE27491AF1C52944FA91 /* BugStatus.swift in Sources */ = {isa = PBXBuildFile; fileRef = 646C676356A4125B8E8D5FA5 /* BugStatus.swift */; }; EEC5853D00ED565270DC867F /* LocalOAuthServer.swift in Sources */ = {isa = PBXBuildFile; fileRef = 34D11523ACB81DBB9D412995 /* LocalOAuthServer.swift */; }; @@ -69,6 +70,7 @@ EF06E1030AA21DA8190169A8 /* KeychainHelper.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = KeychainHelper.swift; sourceTree = ""; }; F1DD178095EFCE7BC121C1C4 /* AppStateService.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = AppStateService.swift; sourceTree = ""; }; 8F2E3D4C5B6A7988061A2B3C /* BitableEventService.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = BitableEventService.swift; sourceTree = ""; }; + 1A2B3C4D5E6F70819293A4B5 /* CalibrationService.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = CalibrationService.swift; sourceTree = ""; }; F888AA5F655343B19839B4C2 /* URL+Feishu.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = "URL+Feishu.swift"; sourceTree = ""; }; FE9FBC39502FA635EF1A6A5C /* FeishuModels.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = FeishuModels.swift; sourceTree = ""; }; /* End PBXFileReference section */ @@ -174,6 +176,7 @@ children = ( F1DD178095EFCE7BC121C1C4 /* AppStateService.swift */, 8F2E3D4C5B6A7988061A2B3C /* BitableEventService.swift */, + 1A2B3C4D5E6F70819293A4B5 /* CalibrationService.swift */, 59CC175DE8E9FE3250757D33 /* NotificationService.swift */, 4F4628E7B599B69B499E09E9 /* PollerService.swift */, 0E7E3775F77DD36C27160A8C /* TokenManager.swift */, @@ -304,6 +307,7 @@ E4CBAE27491AF1C52944FA91 /* BugStatus.swift in Sources */, D2855F82E68FADA8BEA43086 /* AppStateService.swift in Sources */, 7E1F2A3B4C5D6E7F8091A2B3 /* BitableEventService.swift in Sources */, + A1B2C3D4E5F60718293A4B5C /* CalibrationService.swift in Sources */, 4309E8C70AE99466545A3E37 /* FeishuAuthService.swift in Sources */, CB529549DB1DBB5F02884300 /* FeishuError.swift in Sources */, CAE624453A3F0375D872170A /* FeishuModels.swift in Sources */, diff --git a/Sources/AppDelegate.swift b/Sources/AppDelegate.swift index aa0185c..3330251 100644 --- a/Sources/AppDelegate.swift +++ b/Sources/AppDelegate.swift @@ -31,6 +31,7 @@ final class AppDelegate: NSObject, NSApplicationDelegate { // Real-time change push is non-blocking; no-op if the subscribe // base URL isn't configured (falls back to polling). BitableEventService.shared.connect() + CalibrationService.shared.start() } } diff --git a/Sources/Models/AppConfig.swift b/Sources/Models/AppConfig.swift index 012b3fd..1d35218 100644 --- a/Sources/Models/AppConfig.swift +++ b/Sources/Models/AppConfig.swift @@ -21,6 +21,12 @@ struct AppConfig: Codable, Equatable { /// `{feishuAppBaseURL}/api/v1/bitable/events` and refreshes immediately /// on a change push. Empty = disabled (polling only). var feishuAppBaseURL: String = "" + /// When true, Bugger periodically queries bugger-feishu for the list of + /// record IDs assigned to the current user and triggers a fetchNow() if + /// the list doesn't match the bugs in BugStore. Disabled by default — + /// only useful when feishuAppBaseURL is set and push reliability is a + /// concern. + var calibrationEnabled: Bool = false struct FieldMappings: Codable, Equatable { var titleField: String = "Title" @@ -65,7 +71,7 @@ struct AppConfig: Codable, Equatable { case appToken, tableId, assigneeName, fieldMappings case pollIntervalSeconds, dailyRefreshTimes case showFloatingWidget, launchAtLogin, feishuBaseDomain, refreshOnStart - case feishuAppBaseURL + case feishuAppBaseURL, calibrationEnabled } init() {} @@ -83,6 +89,7 @@ struct AppConfig: Codable, Equatable { feishuBaseDomain = try c.decodeIfPresent(String.self, forKey: .feishuBaseDomain) ?? BundledDefault.feishuBaseDomain refreshOnStart = try c.decodeIfPresent(Bool.self, forKey: .refreshOnStart) ?? false feishuAppBaseURL = try c.decodeIfPresent(String.self, forKey: .feishuAppBaseURL) ?? "" + calibrationEnabled = try c.decodeIfPresent(Bool.self, forKey: .calibrationEnabled) ?? false } } diff --git a/Sources/Services/BitableEventService.swift b/Sources/Services/BitableEventService.swift index 3577382..c60d145 100644 --- a/Sources/Services/BitableEventService.swift +++ b/Sources/Services/BitableEventService.swift @@ -22,10 +22,35 @@ final class BitableEventService { /// drop (should auto-reconnect) from an explicit `disconnect()` (should not). private var isEnabled = false - private let reconnectDelay: TimeInterval = 30 + // -- Exponential backoff with jitter ----------------------------------- + + /// Consecutive reconnect attempts since the last successful `connected` + /// event. Reset to 0 when the server confirms the stream is alive. + private var reconnectAttempt = 0 + + /// Base delay (seconds) for the first reconnect attempt. + private let reconnectBaseDelay: TimeInterval = 1 + /// Multiplier applied per attempt. + private let reconnectBackoffFactor: Double = 2 + /// Maximum delay (seconds) — preserves the previous steady-state cadence. + private let reconnectMaxDelay: TimeInterval = 30 + /// Jitter range applied to the computed delay (uniform in [1-range, 1+range]). + private let reconnectJitter: Double = 0.25 private init() {} + /// Compute the reconnect delay for the current attempt count using + /// exponential backoff capped at `reconnectMaxDelay`, then apply ±jitter. + /// Formula: delay = min(cap, base * factor^(attempt-1)) * random(1±jitter) + private func computeReconnectDelay() -> TimeInterval { + let exponent = Double(max(reconnectAttempt, 1) - 1) + let raw = min(reconnectMaxDelay, reconnectBaseDelay * pow(reconnectBackoffFactor, exponent)) + let lo = 1 - reconnectJitter + let hi = 1 + reconnectJitter + let jitterFactor = Double.random(in: lo...hi) + return raw * jitterFactor + } + // MARK: - Public /// Open the SSE connection. Safe to call repeatedly — a no-op if already @@ -97,6 +122,7 @@ final class BitableEventService { session?.invalidateAndCancel() session = nil delegate = nil + reconnectAttempt = 0 } private func handleStreamEnd(error: Error?) { @@ -113,6 +139,10 @@ final class BitableEventService { // Only auto-reconnect if the user still wants to be connected. guard isEnabled else { return } + reconnectAttempt += 1 + let delay = computeReconnectDelay() + BuggerLog.info("BitableEventService: reconnecting in \(String(format: "%.1f", delay))s (attempt \(reconnectAttempt))") + let work = DispatchWorkItem { [weak self] in guard let self, self.isEnabled else { return } guard let url = self.buildURL() else { return } @@ -120,7 +150,7 @@ final class BitableEventService { self.openStream(at: url) } reconnectWorkItem = work - DispatchQueue.main.asyncAfter(deadline: .now() + reconnectDelay, execute: work) + DispatchQueue.main.asyncAfter(deadline: .now() + delay, execute: work) } private func handleEvent(_ event: String, data: String) { @@ -128,9 +158,11 @@ final class BitableEventService { case "change": BuggerLog.info("BitableEventService: change push received (\(data)), refreshing now") Task { await PollerService.shared.fetchNow() } + case "connected": + reconnectAttempt = 0 case "error": BuggerLog.error("BitableEventService: server error \(data)") - case "connected", "heartbeat": + case "heartbeat": break default: BuggerLog.debug("BitableEventService: unknown event \(event)") diff --git a/Sources/Services/CalibrationService.swift b/Sources/Services/CalibrationService.swift new file mode 100644 index 0000000..57f9ab5 --- /dev/null +++ b/Sources/Services/CalibrationService.swift @@ -0,0 +1,124 @@ +import Foundation + +/// Periodically verifies that Bugger's local bug list matches the record IDs +/// the notification server (bugger-feishu) believes are assigned to the +/// current user. If there's a mismatch — meaning a push was missed — this +/// service triggers `PollerService.fetchNow()` to recalibrate. +/// +/// Disabled by default. Enabled via the "Enable calibration check" toggle +/// in Advanced settings. Only active when `feishuAppBaseURL` is configured, +/// since the calibration endpoint lives on the same server as the SSE stream. +final class CalibrationService { + static let shared = CalibrationService() + + private var timer: Timer? + private let interval: TimeInterval = 300 + private var isRunning = false + private let session = URLSession.shared + + private init() {} + + // MARK: - Public + + func start() { + guard !isRunning else { return } + guard let config = AppStateService.shared.config, + config.calibrationEnabled, + !config.feishuAppBaseURL.isEmpty else { return } + isRunning = true + + timer = Timer.scheduledTimer(withTimeInterval: interval, repeats: true) { [weak self] _ in + Task { await self?.calibrate() } + } + timer?.tolerance = 30 + + DispatchQueue.main.asyncAfter(deadline: .now() + 60) { [weak self] in + Task { await self?.calibrate() } + } + + BuggerLog.info("CalibrationService: started (interval=\(Int(interval))s)") + } + + func stop() { + timer?.invalidate() + timer = nil + isRunning = false + } + + func restart() { + stop() + start() + } + + // MARK: - Private + + private func calibrate() async { + guard let config = AppStateService.shared.config, + !config.feishuAppBaseURL.isEmpty, + !config.assigneeName.isEmpty, + !config.fieldMappings.assigneeField.isEmpty else { return } + + guard let url = buildURL(config: config) else { return } + + do { + let (data, response) = try await session.data(from: url) + guard let http = response as? HTTPURLResponse, http.statusCode == 200 else { + BuggerLog.error("CalibrationService: non-200 response (\((response as? HTTPURLResponse)?.statusCode ?? -1))") + return + } + + let result = try JSONDecoder().decode(OwnedRecordsResponse.self, from: data) + + guard result.cacheStatus != "cold" else { + BuggerLog.info("CalibrationService: server cache cold, skipping") + return + } + + let serverIDs = Set(result.recordIds) + let localIDs = Set(BugStore.shared.bugs.map(\.id)) + + guard !localIDs.isEmpty else { + BuggerLog.info("CalibrationService: no local bugs yet, skipping") + return + } + + if serverIDs != localIDs { + let added = serverIDs.subtracting(localIDs) + let removed = localIDs.subtracting(serverIDs) + BuggerLog.info("CalibrationService: mismatch (server=\(serverIDs.count), local=\(localIDs.count), +\(added.count) -\(removed.count)), triggering fetchNow") + await PollerService.shared.fetchNow() + } else { + BuggerLog.debug("CalibrationService: in sync (\(serverIDs.count) records)") + } + } catch { + BuggerLog.error("CalibrationService: error — \(error.localizedDescription)") + } + } + + private func buildURL(config: AppConfig) -> URL? { + let base = config.feishuAppBaseURL.trimmingCharacters(in: CharacterSet(charactersIn: "/")) + guard var comps = URLComponents(string: "\(base)/api/v1/bitable/owned_records") else { + return nil + } + comps.queryItems = [ + URLQueryItem(name: "file_token", value: config.appToken), + URLQueryItem(name: "assignee_field", value: config.fieldMappings.assigneeField), + URLQueryItem(name: "assignee_name", value: config.assigneeName), + ] + return comps.url + } +} + +// MARK: - Response model + +struct OwnedRecordsResponse: Decodable { + let recordIds: [String] + let cacheStatus: String + let count: Int + + enum CodingKeys: String, CodingKey { + case recordIds = "record_ids" + case cacheStatus = "cache_status" + case count + } +} diff --git a/Sources/Services/Feishu/FeishuAuthService.swift b/Sources/Services/Feishu/FeishuAuthService.swift index e79194e..734ebf0 100644 --- a/Sources/Services/Feishu/FeishuAuthService.swift +++ b/Sources/Services/Feishu/FeishuAuthService.swift @@ -80,6 +80,11 @@ final class FeishuAuthService { } func fetchCurrentUserName(accessToken: String) async throws -> String { + let info = try await fetchUserInfo(accessToken: accessToken) + return info.name ?? info.enName ?? "Unknown" + } + + func fetchUserInfo(accessToken: String) async throws -> UserInfoData { let url = URL(string: "\(baseURL)/authen/v1/user_info")! var request = URLRequest(url: url) request.httpMethod = "GET" @@ -88,7 +93,7 @@ final class FeishuAuthService { guard let data = response.data else { throw FeishuError.apiError(code: response.code, message: response.msg) } - return data.name ?? data.enName ?? "Unknown" + return data } private func tenantAccessToken() async throws -> String { diff --git a/Sources/Services/Feishu/FeishuModels.swift b/Sources/Services/Feishu/FeishuModels.swift index 8362fc6..d4fc4d4 100644 --- a/Sources/Services/Feishu/FeishuModels.swift +++ b/Sources/Services/Feishu/FeishuModels.swift @@ -55,6 +55,9 @@ struct UserInfoData: Decodable { let name: String? // en_name → enName via .convertFromSnakeCase let enName: String? + // open_id → openId via .convertFromSnakeCase + // Needed for server-side filtering of Person-type fields on the search endpoint. + let openId: String? } enum JSONValue: Decodable { diff --git a/Sources/Services/Feishu/FeishuService.swift b/Sources/Services/Feishu/FeishuService.swift index b2c3b43..e533b40 100644 --- a/Sources/Services/Feishu/FeishuService.swift +++ b/Sources/Services/Feishu/FeishuService.swift @@ -15,31 +15,16 @@ final class FeishuService { func fetchBugs( config: AppConfig, assigneeName: String, - accessToken: String + accessToken: String, + userOpenId: String? = nil ) async throws -> [Bug] { - var allRecords: [RecordItem] = [] - var pageToken: String? - var pages = 0 - // 25,000 records at 500/page — guard against a runaway cursor that - // would otherwise loop forever accumulating memory. - let maxPages = 50 - - repeat { - pages += 1 - guard pages <= maxPages else { - throw FeishuError.paginationLimitExceeded - } - let page = try await fetchPage( - appToken: config.appToken, - tableId: config.tableId, - pageToken: pageToken, - accessToken: accessToken - ) - allRecords.append(contentsOf: page.items) - pageToken = page.hasMore ? page.pageToken : nil - } while pageToken != nil - - return allRecords + let records = try await fetchRecordsFiltered( + config: config, + assigneeName: assigneeName, + accessToken: accessToken, + userOpenId: userOpenId + ) + return records .map { BugMapper.map($0, config: config) } .filter { bug in bug.assignee.localizedCaseInsensitiveContains(assigneeName) @@ -61,12 +46,118 @@ final class FeishuService { return page.total ?? page.items.count } + // MARK: - Filtered fetch (3-tier fallback) + + /// Fetch records using a 3-tier strategy: + /// 1. Search endpoint with open_id filter (Person-type fields, requires OAuth) + /// 2. List endpoint with CurrentValue filter (text-type fields) + /// 3. Unfiltered list (current behavior — fetch all, filter client-side) + private func fetchRecordsFiltered( + config: AppConfig, + assigneeName: String, + accessToken: String, + userOpenId: String? + ) async throws -> [RecordItem] { + let fieldNames = configFieldNames(config) + + // Tier 1: search endpoint with open_id (Person fields) + if let openId = userOpenId, !openId.isEmpty { + do { + let records = try await searchRecords( + config: config, + accessToken: accessToken, + fieldNames: fieldNames, + assigneeField: config.fieldMappings.assigneeField, + openId: openId + ) + if !records.isEmpty { + BuggerLog.debug("FeishuService: tier 1 (search by open_id) returned \(records.count) records") + return records + } + BuggerLog.debug("FeishuService: tier 1 returned 0 records, falling back") + } catch { + BuggerLog.debug("FeishuService: tier 1 failed (\(error.localizedDescription)), falling back") + } + } + + // Tier 2: list endpoint with CurrentValue filter (text fields) + let filterStr = "CurrentValue.[\(config.fieldMappings.assigneeField)]=\"\(assigneeName)\"" + do { + let records = try await listAllRecords( + config: config, + accessToken: accessToken, + fieldNames: fieldNames, + filter: filterStr + ) + if !records.isEmpty { + BuggerLog.debug("FeishuService: tier 2 (list with filter) returned \(records.count) records") + return records + } + BuggerLog.debug("FeishuService: tier 2 returned 0 records, falling back") + } catch { + BuggerLog.debug("FeishuService: tier 2 failed (\(error.localizedDescription)), falling back") + } + + // Tier 3: unfiltered list (current behavior) + BuggerLog.debug("FeishuService: tier 3 (unfiltered list)") + return try await listAllRecords( + config: config, + accessToken: accessToken, + fieldNames: fieldNames, + filter: nil + ) + } + + /// Build the list of field names to request from the Feishu API, to reduce + /// the per-record payload. Includes all mapped fields. + private func configFieldNames(_ config: AppConfig) -> [String] { + let m = config.fieldMappings + return [m.titleField, m.priorityField, m.statusField, m.assigneeField, + m.reporterField, m.customerField, m.createdAtField, m.updatedAtField] + .filter { !$0.isEmpty } + } + + // MARK: - List endpoint (GET /records) + + private func listAllRecords( + config: AppConfig, + accessToken: String, + fieldNames: [String], + filter: String? + ) async throws -> [RecordItem] { + var allRecords: [RecordItem] = [] + var pageToken: String? + var pages = 0 + let maxPages = 50 + + repeat { + pages += 1 + guard pages <= maxPages else { + throw FeishuError.paginationLimitExceeded + } + let page = try await fetchPage( + appToken: config.appToken, + tableId: config.tableId, + pageToken: pageToken, + accessToken: accessToken, + filter: filter, + fieldNames: fieldNames + ) + allRecords.append(contentsOf: page.items) + pageToken = page.hasMore ? page.pageToken : nil + } while pageToken != nil + + return allRecords + } + private func fetchPage( appToken: String, tableId: String, pageToken: String?, pageSize: Int = 500, - accessToken: String + accessToken: String, + filter: String? = nil, + fieldNames: [String]? = nil ) async throws -> RecordListData { guard !appToken.isEmpty, !tableId.isEmpty else { throw FeishuError.invalidConfiguration("App token and table ID must not be empty.") @@ -80,6 +171,13 @@ final class FeishuService { if let pageToken { queryItems.append(URLQueryItem(name: "page_token", value: pageToken)) } + if let filter, !filter.isEmpty { + queryItems.append(URLQueryItem(name: "filter", value: filter)) + } + if let fieldNames, !fieldNames.isEmpty { + let jsonArray = "[\(fieldNames.map { "\"\($0)\"" }.joined(separator: ","))]" + queryItems.append(URLQueryItem(name: "field_names", value: jsonArray)) + } components.queryItems = queryItems guard let url = components.url else { @@ -117,4 +215,110 @@ final class FeishuService { } return pageData } + + // MARK: - Search endpoint (POST /records/search) + + private func searchRecords( + config: AppConfig, + accessToken: String, + fieldNames: [String], + assigneeField: String, + openId: String + ) async throws -> [RecordItem] { + var allRecords: [RecordItem] = [] + var pageToken: String? + var pages = 0 + let maxPages = 50 + + repeat { + pages += 1 + guard pages <= maxPages else { + throw FeishuError.paginationLimitExceeded + } + let page = try await searchPage( + appToken: config.appToken, + tableId: config.tableId, + pageToken: pageToken, + accessToken: accessToken, + fieldNames: fieldNames, + assigneeField: assigneeField, + openId: openId + ) + allRecords.append(contentsOf: page.items) + pageToken = page.hasMore ? page.pageToken : nil + } while pageToken != nil + + return allRecords + } + + private func searchPage( + appToken: String, + tableId: String, + pageToken: String?, + accessToken: String, + fieldNames: [String], + assigneeField: String, + openId: String + ) async throws -> RecordListData { + guard !appToken.isEmpty, !tableId.isEmpty else { + throw FeishuError.invalidConfiguration("App token and table ID must not be empty.") + } + + let urlString = "\(baseURL)/bitable/v1/apps/\(appToken)/tables/\(tableId)/records/search" + guard let url = URL(string: urlString) else { + throw FeishuError.invalidConfiguration("Could not build search request URL.") + } + + var body: [String: Any] = [ + "page_size": 500, + "filter": [ + "conjunction": "and", + "conditions": [ + [ + "field_name": assigneeField, + "operator": "is", + "value": [openId], + ] + ] + ] + ] + if let pageToken, !pageToken.isEmpty { + body["page_token"] = pageToken + } + if !fieldNames.isEmpty { + body["field_names"] = fieldNames + } + + var request = URLRequest(url: url) + request.httpMethod = "POST" + request.setValue("Bearer \(accessToken)", forHTTPHeaderField: "Authorization") + request.setValue("application/json; charset=utf-8", forHTTPHeaderField: "Content-Type") + request.timeoutInterval = 30 + request.httpBody = try JSONSerialization.data(withJSONObject: body) + + let (data, response) = try await session.data(for: request) + guard let httpResponse = response as? HTTPURLResponse else { + throw FeishuError.networkError(nil) + } + + let rawBody = String(data: data, encoding: .utf8) ?? "" + BuggerLog.debug("searchPage: HTTP \(httpResponse.statusCode) body=\(String(rawBody.prefix(500)))") + + if httpResponse.statusCode == 401 { + throw FeishuError.unauthorized + } + + let apiResponse: FeishuAPIResponse + do { + apiResponse = try decoder.decode(FeishuAPIResponse.self, from: data) + } catch { + BuggerLog.error("searchPage: decode FAILED: \(error)") + throw FeishuError.decodingError(error, rawBody: String(rawBody.prefix(200))) + } + + guard apiResponse.code == 0, let pageData = apiResponse.data else { + throw FeishuError.apiError(code: apiResponse.code, message: apiResponse.msg) + } + return pageData + } } diff --git a/Sources/Services/PollerService.swift b/Sources/Services/PollerService.swift index 0710974..59271c9 100644 --- a/Sources/Services/PollerService.swift +++ b/Sources/Services/PollerService.swift @@ -101,10 +101,12 @@ final class PollerService { let config = try getConfig() let token = try await tokenManager.getAccessToken() let assignee = try await tokenManager.resolveAssigneeName(config: config, accessToken: token) + let openId = tokenManager.cachedOpenId let bugs = try await feishuService.fetchBugs( config: config, assigneeName: assignee, - accessToken: token + accessToken: token, + userOpenId: openId ) await MainActor.run { diff --git a/Sources/Services/TokenManager.swift b/Sources/Services/TokenManager.swift index 59caa62..826a924 100644 --- a/Sources/Services/TokenManager.swift +++ b/Sources/Services/TokenManager.swift @@ -10,6 +10,7 @@ final class TokenManager { private let accessTokenKey = "feishu.access_token" private let refreshTokenKey = "feishu.refresh_token" private let tokenExpiryKey = "feishu.token_expiry" + private let openIdKey = "feishu.open_id" enum State: Equatable { case unauthenticated @@ -38,6 +39,18 @@ final class TokenManager { return authService.authorizeURL } + /// The user's Feishu open_id, cached after the first OAuth callback or + /// user-info fetch. Used for server-side filtering of Person-type fields + /// on the Feishu search endpoint. Nil if the user hasn't authenticated. + var cachedOpenId: String? { + keychain.read(openIdKey) + } + + func storeOpenId(_ openId: String) { + guard !openId.isEmpty else { return } + keychain.write(openId, forKey: openIdKey) + } + func getAccessToken() async throws -> String { if let token = keychain.read(accessTokenKey), let expiry = UserDefaults.standard.object(forKey: tokenExpiryKey) as? Date, @@ -83,6 +96,7 @@ final class TokenManager { func clearTokens() { keychain.delete(accessTokenKey) keychain.delete(refreshTokenKey) + keychain.delete(openIdKey) UserDefaults.standard.removeObject(forKey: tokenExpiryKey) state = .unauthenticated } @@ -105,6 +119,11 @@ final class TokenManager { refresh: response.refreshToken ?? "", expiresIn: response.expiresIn ) + // Fetch and cache the user's open_id for server-side record filtering. + if let userInfo = try? await authService.fetchUserInfo(accessToken: response.accessToken), + let openId = userInfo.openId { + storeOpenId(openId) + } state = .authenticated BuggerLog.info("handleCallback: done, authenticated ✓") } @@ -113,7 +132,11 @@ final class TokenManager { if !config.assigneeName.isEmpty { return config.assigneeName } - return try await authService.fetchCurrentUserName(accessToken: accessToken) + let userInfo = try await authService.fetchUserInfo(accessToken: accessToken) + if let openId = userInfo.openId { + storeOpenId(openId) + } + return userInfo.name ?? userInfo.enName ?? "Unknown" } } diff --git a/Sources/Views/Settings/SettingsView.swift b/Sources/Views/Settings/SettingsView.swift index 44a2104..a25a834 100644 --- a/Sources/Views/Settings/SettingsView.swift +++ b/Sources/Views/Settings/SettingsView.swift @@ -139,6 +139,16 @@ struct SettingsView: View { Section { DisclosureGroup(isExpanded: $showAdvanced) { + VStack(alignment: .leading, spacing: 4) { + Toggle("Enable calibration check", isOn: $config.calibrationEnabled) + Text("Periodically verifies with the notification server that all your assigned bugs are in sync. Helps recover missed push notifications. Only works when a Subscribe base URL is set.") + .font(.caption) + .foregroundStyle(.secondary) + .fixedSize(horizontal: false, vertical: true) + } + + Divider() + Text("Field Mappings") .font(.caption) .fontWeight(.semibold) @@ -179,6 +189,7 @@ struct SettingsView: View { TokenManager.shared.clearTokens() PollerService.shared.stop() BitableEventService.shared.disconnect() + CalibrationService.shared.stop() } .foregroundStyle(.red) } @@ -291,6 +302,7 @@ struct SettingsView: View { // connect() is a no-op when the URL is empty, so clearing the field // disables real-time push. BitableEventService.shared.reconnect() + CalibrationService.shared.restart() } } diff --git a/docs/PUSH_RELIABILITY_PLAN.md b/docs/PUSH_RELIABILITY_PLAN.md new file mode 100644 index 0000000..b4ff7ea --- /dev/null +++ b/docs/PUSH_RELIABILITY_PLAN.md @@ -0,0 +1,724 @@ +# Push Reliability — Design & Implementation Plan + +**Date:** 2026-07-10 +**Status:** Draft — pending review +**Scope:** Bugger (macOS SSE client) + bugger-feishu (k8s SSE server) + +--- + +## 1. Problem Statement + +Push notifications (SSE `change` events from bugger-feishu → Bugger) can be +**delayed or missed entirely**. The root cause is unknown — it could be: + +- **Feishu side** failing to deliver WebSocket events to bugger-feishu. +- **bugger-feishu** failing to push the SSE event to the client (e.g. connection + dropped, queue overflow, server crash). +- **Ingress** (k8s nginx / cloud LB) silently closing the long-lived SSE + connection before the heartbeat resets the idle timer (see + `SSE_TIMEOUT_ANALYSIS.md`). + +The current reconnection strategy (fixed 30s delay, no jitter) compounds the +problem: when the ingress drops many clients simultaneously, they all retry at +the same instant — a thundering herd that can overwhelm the ingress and prevent +any client from reconnecting. + +### Goals + +| # | Goal | Owner | +|---|------|-------| +| 1 | Reconnect with **exponential backoff + jitter** to avoid retry storms | Bugger | +| 2 | bugger-feishu exposes an **owned-records endpoint** so Bugger can calibrate | bugger-feishu | +| 3 | Bugger has an **advanced settings toggle** (disabled by default) to enable periodic calibration | Bugger | +| 4 | Bugger **optimizes the Feishu query** — fetch only the owner's bugs, not all records | Bugger | + +--- + +## 2. Architecture Overview + +``` +┌──────────────────────────────────────────────────────────────────────┐ +│ Bugger (macOS) │ +│ │ +│ ┌──────────────────┐ SSE push ┌──────────────────────────────┐ │ +│ │ BitableEventSvc │ ◀─────────────│ Ingress → bugger-feishu │ │ +│ │ (exponential │ change │ (SSE server, 30s heartbeat) │ │ +│ │ backoff+jitter)│ └──────────────────────────────┘ │ +│ └────────┬─────────┘ │ │ +│ │ fetchNow() │ GET /owned_records │ +│ ▼ ▼ │ +│ ┌──────────────────┐ ┌──────────────────────┐ │ +│ │ PollerService │ │ CalibrationService │ │ +│ │ (periodic poll │ │ (optional, every │ │ +│ │ + fetchNow) │ │ 5 min, compares │ │ +│ └────────┬─────────┘ │ record IDs) │ │ +│ │ └──────────┬───────────┘ │ +│ ▼ │ mismatch? │ +│ ┌──────────────────┐ │ │ +│ │ FeishuService │ ◀─────────────────────┘ fetchNow() │ +│ │ (search records │ │ +│ │ with filter) │ │ +│ └──────────────────┘ │ +└──────────────────────────────────────────────────────────────────────┘ +``` + +**Data flow for calibration:** +1. `CalibrationService` calls `GET {feishuAppBaseURL}/api/v1/bitable/owned_records` +2. bugger-feishu returns the record IDs assigned to this user (from its in-memory cache) +3. Bugger compares the returned IDs with `BugStore.shared.bugs.map(\.id)` +4. On mismatch → triggers `PollerService.shared.fetchNow()` → `FeishuService` fetches + only the owner's bugs (server-side filtered) → `BugStore` updates + +--- + +## 3. Design Details + +### 3.1 Item 1 — Exponential Backoff with Jitter + +**Problem:** `BitableEventService.swift:25` uses a fixed `reconnectDelay = 30`. +All clients retry on the same cadence, creating synchronized load spikes. + +**Design:** + +Replace the fixed delay with exponential backoff + decorrelated jitter: + +``` +attempt 1: delay ≈ 1s (±25% jitter) +attempt 2: delay ≈ 2s (±25% jitter) +attempt 3: delay ≈ 4s (±25% jitter) +attempt 4: delay ≈ 8s (±25% jitter) +attempt 5: delay ≈ 16s (±25% jitter) +attempt 6+: delay ≈ 30s (±25% jitter) ← cap +``` + +**Formula:** +``` +base = 1.0 (seconds) +factor = 2.0 +cap = 30.0 (seconds) — preserves current steady-state behavior +jitter = ±25% (random uniform in [0.75, 1.25]) + +raw_delay = min(cap, base * factor^(attempt - 1)) +delay = raw_delay * Double.random(in: 0.75...1.25) +``` + +**State management:** +- `reconnectAttempt: Int = 0` — incremented on each `handleStreamEnd`, reset to 0 + when a `connected` event is received from the server. +- The attempt counter is an instance variable on `BitableEventService`, mutated + only on the main queue (all callbacks are already routed to main). + +**Reset logic:** +In `handleEvent`, when event == `"connected"`: +```swift +reconnectAttempt = 0 +``` +This ensures the backoff de-escalates as soon as the connection is healthy. + +**No catch-up fetch on reconnect (unchanged):** +The existing `handleEvent` already calls `PollerService.shared.fetchNow()` on +`change` events. The calibration service (item 3) provides the safety net for +missed events, so we do not add an explicit catch-up fetch on reconnect. + +**Changes:** +| File | Change | +|------|--------| +| `bugger/Sources/Services/BitableEventService.swift` | Replace `reconnectDelay` constant with backoff logic; add `reconnectAttempt` counter; reset on `connected` event | + +--- + +### 3.2 Item 2 — bugger-feishu Owned-Records Endpoint + +**Problem:** Bugger has no way to verify it has received all change pushes. +If a push is missed (SSE connection down, Feishu WS event lost), the bug list +is stale until the next periodic poll. + +**Design:** + +New endpoint on bugger-feishu: + +``` +GET /api/v1/bitable/owned_records + ?file_token={app_token} + &assignee_field={field_name} + &assignee_name={user_name} +``` + +**Response (200):** +```json +{ + "file_token": "bascnXXX", + "assignee_name": "张三", + "assignee_field": "负责人", + "record_ids": ["recAAA", "recBBB", "recCCC"], + "count": 3, + "cache_status": "warm", + "server_time": 1720000000 +} +``` + +**`cache_status` values:** +| Value | Meaning | Bugger action | +|-------|---------|---------------| +| `"warm"` | Cache has been populated (WS events received or full pull done) | Compare IDs, calibrate on mismatch | +| `"cold"` | Cache is empty — no WS events received yet, no full pull done | **Skip calibration** — would cause false positives | +| `"stale"` | Cache exists but hasn't been refreshed recently (> 1h since last update) | Compare IDs, but log a warning | + +**Implementation on bugger-feishu:** + +The server already maintains `self._cache: Dict[str, Dict[str, Dict[str, str]]]` +(file_token → assignee_field → record_id → assignee_value) in +`BitableEventService`. The endpoint reads from this cache — **no Feishu API call** +is needed, making it fast and cheap. + +```python +# bitable_event_service.py — new method + +def get_owned_record_ids( + self, file_token: str, assignee_field: str, assignee_name: str +) -> dict: + """Return the record IDs assigned to a user, from the in-memory cache.""" + file_cache = self._cache.get(file_token, {}) + field_cache = file_cache.get(assignee_field, {}) + + if not field_cache: + # Check if we've ever seen this file_token at all + if file_token not in self._cache: + return {"record_ids": [], "cache_status": "cold", "count": 0} + return {"record_ids": [], "cache_status": "cold", "count": 0} + + # Check staleness: if no table_id learned yet, cache may be incomplete + has_table_id = file_token in self._table_ids + + owned = [ + rid for rid, assignee in field_cache.items() + if assignee == assignee_name + ] + status = "warm" if has_table_id else "stale" + return { + "record_ids": owned, + "cache_status": status, + "count": len(owned), + } +``` + +```python +# bitable_subscription.py — new route + +@router.get("/owned_records") +async def owned_records( + file_token: str = Query(...), + assignee_field: str = Query(...), + assignee_name: str = Query(...), +): + service = getattr(request.app.state, "bitable_event_service", None) + if service is None: + return {"record_ids": [], "cache_status": "cold", "count": 0} + + result = service.get_owned_record_ids(file_token, assignee_field, assignee_name) + result["file_token"] = file_token + result["assignee_name"] = assignee_name + result["assignee_field"] = assignee_field + result["server_time"] = int(time.time()) + return result +``` + +**Why read from cache, not live Feishu API?** +- The cache is already maintained and reconciled every 24h by the server. +- A live API call per calibration request (every 5 min × N clients) would be + expensive and rate-limited. +- The cache is "good enough" for calibration — it catches missed pushes, which + is the goal. The periodic reconciliation task (every 24h) ensures the cache + stays eventually consistent. + +**Changes:** +| File | Change | +|------|--------| +| `bugger-feishu/app/services/bitable_event_service.py` | Add `get_owned_record_ids()` method | +| `bugger-feishu/app/api/bitable_subscription.py` | Add `GET /owned_records` route | +| `bugger-feishu/tests/test_bitable_change_notification.py` | Add unit tests for the endpoint | + +--- + +### 3.3 Item 3 — Bugger Calibration Service + Advanced Settings Toggle + +**Problem:** Even with backoff, the SSE connection may be unreliable (ingress +issues, server restarts). A periodic calibration check provides a safety net. + +**Design:** + +#### 3.3.1 AppConfig — new field + +```swift +// AppConfig.swift +/// When true, Bugger periodically queries bugger-feishu for the list of +/// record IDs assigned to the current user and triggers a fetchNow() if +/// the list doesn't match the bugs in BugStore. Disabled by default — +/// only useful when feishuAppBaseURL is set and push reliability is a +/// concern. +var calibrationEnabled: Bool = false +``` + +- Default: `false` (disabled) +- Only meaningful when `feishuAppBaseURL` is set (the calibration endpoint + lives on the same server) +- Tolerant decoding (like all other AppConfig fields) so existing persisted + configs don't break + +#### 3.3.2 CalibrationService — new service + +```swift +// Sources/Services/CalibrationService.swift + +final class CalibrationService { + static let shared = CalibrationService() + + private var timer: Timer? + private let interval: TimeInterval = 300 // 5 minutes + private var isRunning = false + private let session = URLSession.shared + + private init() {} + + func start() { + guard !isRunning else { return } + guard let config = AppStateService.shared.config, + config.calibrationEnabled, + !config.feishuAppBaseURL.isEmpty else { return } + isRunning = true + + // First check after 60s (let the initial fetch + SSE connect settle) + timer = Timer.scheduledTimer(withTimeInterval: interval, repeats: true) { [weak self] _ in + Task { await self?.calibrate() } + } + timer?.tolerance = 30 + // Delay the first run + DispatchQueue.main.asyncAfter(deadline: .now() + 60) { [weak self] in + Task { await self?.calibrate() } + } + } + + func stop() { + timer?.invalidate() + timer = nil + isRunning = false + } + + func restart() { + stop() + start() + } + + private func calibrate() async { + guard let config = AppStateService.shared.config, + !config.feishuAppBaseURL.isEmpty, + !config.assigneeName.isEmpty, + !config.fieldMappings.assigneeField.isEmpty else { return } + + // Build URL: {feishuAppBaseURL}/api/v1/bitable/owned_records?... + guard let url = buildURL(config: config) else { return } + + do { + let (data, response) = try await session.data(from: url) + guard let http = response as? HTTPURLResponse, http.statusCode == 200 else { return } + + let result = try JSONDecoder().decode(OwnedRecordsResponse.self, from: data) + + // Skip if cache is cold — would cause false positives + guard result.cacheStatus == "warm" || result.cacheStatus == "stale" else { return } + + let serverIDs = Set(result.recordIds) + let localIDs = Set(BugStore.shared.bugs.map(\.id)) + + // Only calibrate if we have local data (skip on fresh launch + // before the first fetch completes) + guard !localIDs.isEmpty else { return } + + let mismatch = serverIDs != localIDs + if mismatch { + BuggerLog.info("CalibrationService: mismatch detected (server=\(serverIDs.count), local=\(localIDs.count)), triggering fetchNow") + await PollerService.shared.fetchNow() + } + } catch { + BuggerLog.error("CalibrationService: error — \(error.localizedDescription)") + } + } + + private func buildURL(config: AppConfig) -> URL? { + let base = config.feishuAppBaseURL.trimmingCharacters(in: CharacterSet(charactersIn: "/")) + guard var comps = URLComponents(string: "\(base)/api/v1/bitable/owned_records") else { return nil } + comps.queryItems = [ + URLQueryItem(name: "file_token", value: config.appToken), + URLQueryItem(name: "assignee_field", value: config.fieldMappings.assigneeField), + URLQueryItem(name: "assignee_name", value: config.assigneeName), + ] + return comps.url + } +} + +// Response model +struct OwnedRecordsResponse: Decodable { + let recordIds: [String] + let cacheStatus: String + let count: Int + + enum CodingKeys: String, CodingKey { + case recordIds = "record_ids" + case cacheStatus = "cache_status" + case count + } +} +``` + +#### 3.3.3 Settings UI — advanced section + +In `SettingsView.swift`, inside the existing `DisclosureGroup` for Advanced +options, add a new section before Field Mappings: + +```swift +// Advanced section — calibration toggle +VStack(alignment: .leading, spacing: 4) { + Toggle("Enable calibration check", isOn: $config.calibrationEnabled) + Text("Periodically verifies with the notification server that all your assigned bugs are in sync. Helps recover missed push notifications. Only works when a Subscribe base URL is set.") + .font(.caption) + .foregroundStyle(.secondary) +} +``` + +#### 3.3.4 Lifecycle wiring + +| Location | Action | +|----------|--------| +| `AppDelegate.applicationDidFinishLaunching` | After `BitableEventService.shared.connect()`, add `CalibrationService.shared.start()` | +| `SettingsView.save()` | After `BitableEventService.shared.reconnect()`, add `CalibrationService.shared.restart()` | +| `SettingsView` Disconnect button | Add `CalibrationService.shared.stop()` | + +**Changes:** +| File | Change | +|------|--------| +| `bugger/Sources/Models/AppConfig.swift` | Add `calibrationEnabled` field + CodingKey + tolerant decode | +| `bugger/Sources/Services/CalibrationService.swift` | **New file** — calibration service | +| `bugger/Sources/Views/Settings/SettingsView.swift` | Add toggle in Advanced section; wire lifecycle in `save()` and Disconnect | +| `bugger/Sources/AppDelegate.swift` | Start calibration on launch | + +--- + +### 3.4 Item 4 — Optimize Feishu Query (Filter by Assignee) + +**Problem:** `FeishuService.fetchBugs()` fetches **ALL** records from the Bitable +table (up to 25,000, paginated 500/page = 50 API calls), then filters +client-side by assignee name. For large tables, this is slow and wasteful. + +**Feishu API capabilities (verified from official docs):** + +Two approaches are available: + +| Approach | Endpoint | Filter format | Person field support | +|----------|----------|---------------|---------------------| +| **A. List records** | `GET .../records` | `filter` query param: `CurrentValue.[FieldName]="value"` | Unreliable for Person fields — test code already notes it may not work | +| **B. Search records** | `POST .../records/search` | JSON body with structured `filter` object | **Supported** — Person fields require `open_id` (user ID), not display name | + +**Key constraint (from Feishu filter guide):** +> Person field value: `["ou_9a971ded01b4ca66f4798549878abcef"]` — Fill in the +> corresponding user **ID**. The user ID type must match the `user_id_type` +> parameter (default: `open_id`). + +This means filtering by Person fields requires the user's `open_id`, which is +only available if the user authenticated via OAuth. If the user entered their +name manually in Settings (no OAuth), we cannot filter Person fields server-side. + +**Chosen strategy — multi-tier with fallback:** + +``` +1. If user's open_id is available (from OAuth): + → Use search endpoint with filter: {field_name, operator: "is", value: [open_id]} + → Works for Person-type assignee fields + +2. If open_id not available OR search returns error/0 results: + → Use list endpoint with filter: CurrentValue.[assigneeField]="assigneeName" + → Works for text-type assignee fields + +3. If both fail: + → Fall back to current unfiltered list endpoint (fetch all, filter client-side) +``` + +**Additionally — `field_names` optimization (applies to all tiers):** + +Both list and search endpoints support requesting only specific fields. By +passing only the fields Bugger needs (title, priority, status, assignee, reporter, +customer, created_at, updated_at), we reduce the per-record payload significantly +— especially in tables with many columns or large text/attachment fields. + +#### 3.4.1 Capture user's open_id during OAuth + +Extend `UserInfoData` to include `openId`: + +```swift +// FeishuModels.swift +struct UserInfoData: Decodable { + let name: String? + let enName: String? + let openId: String? // NEW — needed for server-side Person field filtering +} +``` + +Extend `TokenManager` to cache the open_id: + +```swift +// TokenManager.swift +private let openIdKey = "feishu.open_id" + +var cachedOpenId: String? { + keychain.read(openIdKey) +} + +// In handleCallback() and refreshAccessToken() — after getting user info: +// Store open_id alongside tokens +``` + +#### 3.4.2 New FeishuService method — search records with filter + +```swift +// FeishuService.swift + +func fetchBugs( + config: AppConfig, + assigneeName: String, + accessToken: String, + userOpenId: String? = nil // NEW parameter +) async throws -> [Bug] { + // Try filtered search first, fall back to unfiltered list + let records = try await fetchRecordsFiltered( + config: config, assigneeName: assigneeName, + accessToken: accessToken, userOpenId: userOpenId + ) + return records.map { BugMapper.map($0, config: config) } + .filter { bug in bug.assignee.localizedCaseInsensitiveContains(assigneeName) } +} + +private func fetchRecordsFiltered( + config: AppConfig, assigneeName: String, + accessToken: String, userOpenId: String? +) async throws -> [RecordItem] { + // Tier 1: search endpoint with open_id filter (Person fields) + if let openId = userOpenId { + if let records = try? await searchRecords( + config: config, accessToken: accessToken, + filterConditions: [ + ["field_name": config.fieldMappings.assigneeField, + "operator": "is", + "value": [openId]] + ] + ), !records.isEmpty { + return records + } + } + + // Tier 2: list endpoint with CurrentValue filter (text fields) + let filterStr = "CurrentValue.[\(config.fieldMappings.assigneeField)]=\"\(assigneeName)\"" + if let records = try? await listRecords( + config: config, accessToken: accessToken, filter: filterStr + ), !records.isEmpty { + return records + } + + // Tier 3: fallback — unfiltered list (current behavior) + return try await listRecords(config: config, accessToken: accessToken, filter: nil) +} +``` + +**Note:** The client-side `.filter` on `assigneeName` is retained as a safety +net even after server-side filtering. It's a no-op if the filter worked +correctly, but prevents incorrect records from appearing if the filter is +imperfect. + +#### 3.4.3 PollerService — pass open_id to FeishuService + +```swift +// PollerService.swift — in performFetch() +let openId = TokenManager.shared.cachedOpenId +let bugs = try await feishuService.fetchBugs( + config: config, + assigneeName: assignee, + accessToken: token, + userOpenId: openId // NEW +) +``` + +**Changes:** +| File | Change | +|------|--------| +| `bugger/Sources/Services/Feishu/FeishuModels.swift` | Add `openId` to `UserInfoData` | +| `bugger/Sources/Services/TokenManager.swift` | Cache `open_id` from OAuth; expose `cachedOpenId` | +| `bugger/Sources/Services/Feishu/FeishuService.swift` | Add `searchRecords` method; add `fetchRecordsFiltered` with 3-tier fallback; add `field_names` to list calls; accept `userOpenId` parameter | +| `bugger/Sources/Services/PollerService.swift` | Pass `userOpenId` to `fetchBugs` | + +--- + +## 4. Ingress Considerations + +The user suspects the ingress plays a role in missed pushes. The code changes +above make Bugger **resilient** to ingress issues, but the ingress itself should +also be configured correctly (per `SSE_TIMEOUT_ANALYSIS.md` R1): + +| Setting | Required value | Why | +|---------|---------------|-----| +| `proxy_read_timeout` | ≥ 60s | Must exceed the 30s heartbeat with margin | +| `proxy_buffering` | `off` | Prevents heartbeats from being buffered and never sent | +| `proxy_send_timeout` | ≥ 60s | Symmetric to read timeout | +| Cloud LB idle-timeout | ≥ 60s | Some LBs override ingress settings | + +**How the code changes mitigate ingress issues:** +- **Backoff + jitter** → When the ingress drops many SSE connections (e.g. on + pod restart), clients reconnect at randomized intervals instead of + simultaneously, reducing the load spike that could cause the ingress to reject + new connections. +- **Calibration service** → Even if the SSE connection is chronically + unreliable (ingress kills it every cycle), the 5-minute calibration check + ensures the bug list stays eventually consistent. +- **Optimized Feishu query** → When calibration triggers a `fetchNow()`, the + filtered query is much faster and lighter, reducing the load on both the + Feishu API and the client. + +--- + +## 5. Implementation Plan + +### Phase 1 — Bugger: Exponential Backoff (Item 1) + +**Scope:** `BitableEventService.swift` only. No server changes. No new files. + +| Step | Task | File | +|------|------|------| +| 1.1 | Remove `reconnectDelay` constant; add `reconnectAttempt`, `reconnectBaseDelay`, `reconnectMaxDelay`, `reconnectJitterRange` properties | `BitableEventService.swift` | +| 1.2 | Add `computeReconnectDelay() -> TimeInterval` method implementing the backoff + jitter formula | `BitableEventService.swift` | +| 1.3 | Update `handleStreamEnd` to call `computeReconnectDelay()` and increment `reconnectAttempt` | `BitableEventService.swift` | +| 1.4 | Update `handleEvent` to reset `reconnectAttempt = 0` on `"connected"` event | `BitableEventService.swift` | +| 1.5 | Reset `reconnectAttempt = 0` in `closeStream()` (clean disconnect resets the counter) | `BitableEventService.swift` | + +**Testing:** Manual — disconnect Wi-Fi, observe reconnect delays in logs. +Verify: 1st retry ~1s, 2nd ~2s, 3rd ~4s... capping at ~30s. Reconnect after +recovery resets to 1s. + +### Phase 2 — bugger-feishu: Owned-Records Endpoint (Item 2) + +**Scope:** bugger-feishu server only. + +| Step | Task | File | +|------|------|------| +| 2.1 | Add `get_owned_record_ids()` method to `BitableEventService` | `bitable_event_service.py` | +| 2.2 | Add `GET /owned_records` route to `bitable_subscription.py` | `bitable_subscription.py` | +| 2.3 | Add unit tests (mock cache warm/cold/stale states) | `test_bitable_change_notification.py` | + +**Testing:** `pytest bugger-feishu/tests/test_bitable_change_notification.py` + +### Phase 3 — Bugger: Calibration Service + Settings (Item 3) + +**Scope:** Bugger client only. Depends on Phase 2 (endpoint must exist). + +| Step | Task | File | +|------|------|------| +| 3.1 | Add `calibrationEnabled: Bool = false` to `AppConfig` (field + CodingKey + tolerant decode) | `AppConfig.swift` | +| 3.2 | Create `CalibrationService.swift` — timer, `calibrate()`, `OwnedRecordsResponse` model | **New file** | +| 3.3 | Add calibration toggle in Advanced section of `SettingsView` | `SettingsView.swift` | +| 3.4 | Wire `CalibrationService.shared.start()` in `AppDelegate` | `AppDelegate.swift` | +| 3.5 | Wire `CalibrationService.shared.restart()` in `SettingsView.save()` | `SettingsView.swift` | +| 3.6 | Wire `CalibrationService.shared.stop()` in Disconnect button | `SettingsView.swift` | + +**Testing:** Manual — enable calibration in Advanced settings, verify the timer +fires every 5 min (check logs), verify `fetchNow()` triggers on ID mismatch. + +### Phase 4 — Bugger: Optimized Feishu Query (Item 4) + +**Scope:** Bugger client only. Independent of Phases 2-3. + +| Step | Task | File | +|------|------|------| +| 4.1 | Add `openId` to `UserInfoData` | `FeishuModels.swift` | +| 4.2 | Cache `open_id` in `TokenManager` (store on OAuth callback + token refresh) | `TokenManager.swift` | +| 4.3 | Add `searchRecords()` method using `POST .../records/search` with structured filter | `FeishuService.swift` | +| 4.4 | Add `fetchRecordsFiltered()` with 3-tier fallback (open_id → name → unfiltered) | `FeishuService.swift` | +| 4.5 | Add `field_names` parameter to `listRecords`/`searchRecords` to reduce payload | `FeishuService.swift` | +| 4.6 | Update `PollerService.performFetch()` to pass `userOpenId` | `PollerService.swift` | + +**Testing:** Manual — verify filtered query returns correct results (check +`BuggerLog.debug` output for which tier was used). Test with both Person-type +and text-type assignee fields. Verify fallback works when open_id is unavailable. + +### Phase 5 — Verification & Documentation + +| Step | Task | +|------|------| +| 5.1 | Build Bugger in Xcode — verify no compiler errors | +| 5.2 | Run bugger-feishu tests — `pytest` | +| 5.3 | End-to-end test: start bugger-feishu → connect Bugger → kill SSE → verify backoff → verify calibration catches missed push | +| 5.4 | Update `SSE_TIMEOUT_ANALYSIS.md` — mark R3 (backoff) as implemented, link to this plan | + +--- + +## 6. Key Code Locations (current state) + +All paths relative to `/Users/tigeren/Dev/aptsell/bugger-boundle/`: + +### Bugger (macOS client) + +| Location | What | +|----------|------| +| `bugger/Sources/Services/BitableEventService.swift:25` | `reconnectDelay = 30` (fixed — to be replaced with backoff) | +| `bugger/Sources/Services/BitableEventService.swift:102-124` | `handleStreamEnd` — reconnect scheduling | +| `bugger/Sources/Services/BitableEventService.swift:126-138` | `handleEvent` — `connected` event (no-op — to reset attempt counter) | +| `bugger/Sources/Services/BitableEventService.swift:140-165` | `buildURL` — SSE endpoint URL construction | +| `bugger/Sources/Services/PollerService.swift:88-126` | `performFetch` — calls `FeishuService.fetchBugs` | +| `bugger/Sources/Services/Feishu/FeishuService.swift:15-47` | `fetchBugs` — fetches ALL records, filters client-side | +| `bugger/Sources/Services/Feishu/FeishuService.swift:64-119` | `fetchPage` — list records API call | +| `bugger/Sources/Services/Feishu/FeishuModels.swift:54-58` | `UserInfoData` — has `name`, `enName` (needs `openId`) | +| `bugger/Sources/Services/TokenManager.swift:112-117` | `resolveAssigneeName` — returns name, not open_id | +| `bugger/Sources/Models/AppConfig.swift:3-87` | `AppConfig` — needs `calibrationEnabled` field | +| `bugger/Sources/Views/Settings/SettingsView.swift:140-170` | Advanced section — needs calibration toggle | +| `bugger/Sources/AppDelegate.swift:29-34` | App launch — needs `CalibrationService.start()` | + +### bugger-feishu (SSE server) + +| Location | What | +|----------|------| +| `bugger-feishu/app/api/bitable_subscription.py:24-94` | SSE `/events` endpoint — model for new `/owned_records` route | +| `bugger-feishu/app/services/bitable_event_service.py:49-66` | `BitableEventService` — maintains `_cache` (record_id → assignee) | +| `bugger-feishu/app/services/bitable_event_service.py:124-182` | `warm_cache` — full pull that populates the cache | +| `bugger-feishu/app/services/bitable_event_service.py:215-263` | `add_connection` / `remove_connection` — SSE lifecycle | +| `bugger-feishu/app/services/bitable_event_service.py:460-525` | Reconciliation task — periodic full pull (24h) | +| `bugger-feishu/app/main.py:88-91` | Router registration — add `/owned_records` route here | + +--- + +## 7. Risks & Mitigations + +| Risk | Impact | Mitigation | +|------|--------|------------| +| Feishu search endpoint filter doesn't work for Person fields with open_id | Filtered query returns 0 results → fallback to unfiltered (current behavior) | 3-tier fallback strategy ensures no regression | +| Calibration false positives (server cache stale, triggers unnecessary fetchNow) | Extra Feishu API calls every 5 min | `cache_status` check — skip when cold; `localIDs.isEmpty` check — skip on fresh launch; 5-min interval limits impact | +| Server cache doesn't include newly assigned records (reconciliation runs every 24h) | Calibration misses recent assignments | The SSE push should catch these; calibration is a safety net, not the primary mechanism. Server reconciliation interval can be reduced if needed. | +| Backoff delays push recovery (30s cap vs. current 30s fixed) | Slightly slower first reconnect (1s vs 30s — actually faster!) | Backoff starts at 1s (faster than current 30s), caps at 30s. Net improvement. | +| `open_id` not available (user entered name manually, no OAuth) | Can't filter Person fields server-side | Fall back to text filter or unfiltered list. No regression. | + +--- + +## 8. Open Questions + +1. **Calibration interval** — 5 min is proposed. Should it be configurable in + settings, or fixed? (Recommend: fixed for simplicity; can add a setting + later if needed.) + +2. **Server cache freshness** — The server's reconciliation runs every 24h. + Should the calibration endpoint trigger a cache refresh if the cache is + stale? (Recommend: no — keep the endpoint read-only and cheap. The + reconciliation task handles freshness.) + +3. **Person field filter with `contains` operator** — Could we use + `operator: "contains"` with the user's name instead of `is` with open_id? + The Feishu docs say Person field values must be user IDs, but `contains` + might behave differently. (Recommend: test during implementation; if + `contains` with name works, it simplifies the flow by not requiring open_id.) + +4. **`field_names` URL encoding** — The list endpoint expects `field_names` as + a JSON array string (e.g. `["Title","Status"]`). Need to verify that + `URLQueryItem` encodes this correctly for the Feishu API. diff --git a/docs/SSE_TIMEOUT_ANALYSIS.md b/docs/SSE_TIMEOUT_ANALYSIS.md new file mode 100644 index 0000000..a8e5142 --- /dev/null +++ b/docs/SSE_TIMEOUT_ANALYSIS.md @@ -0,0 +1,302 @@ +# SSE Long-Lived Connection — Timeout & Reconnection Analysis + +**Date:** 2026-07-10 +**Status:** Analysis complete — recommendations pending implementation +**Scope:** Bugger (macOS SSE client) → Ingress → bugger-feishu (k8s pod, SSE server) + +--- + +## TL;DR + +Bugger subscribes to bugger-feishu via a long-lived SSE connection that traverses a +Kubernetes ingress. The connection's survival depends on **server-side heartbeats** and +the **ingress idle-timeout configuration** — neither of which Bugger can control from the +client side. + +| Concern | Finding | Risk | +|---|---|---| +| Heartbeat | Server emits `: heartbeat\n\n` every ~30s; client relies on it | ✅ if ingress idle-timeout ≥ 60s; ❌ if tighter | +| Reconnection | Present, fixed 30s delay, well-guarded | Works, but no backoff/jitter → retry storms | +| Exponential backoff | **Not implemented** | Retry storms under sustained outage | + +**Top recommendations:** +1. Set ingress `proxy_read_timeout` / LB idle-timeout to **≥ 60s** with `proxy_buffering off`. +2. Verify the 30s heartbeat is actually deployed in bugger-feishu (it is only *specified* in + this repo's docs, not in shipped code). +3. Add exponential backoff with jitter and a cap to `BitableEventService` reconnect logic. + +--- + +## 1. Architecture Under Test + +``` +┌──────────────┐ SSE (outbound, long-lived) ┌─────────────────────────────┐ +│ Bugger │ ──────────────────────────────▶ │ Kubernetes Ingress │ +│ (macOS) │ GET /api/v1/bitable/events │ (idle-timeout = T_ingress) │ +│ SSE client │ Accept: text/event-stream │ │ │ +└──────────────┘ │ ▼ │ + │ ┌────────────────────────┐ │ + │ │ bugger-feishu pod │ │ + │ │ (FastAPI SSE server) │ │ + │ │ heartbeat every 30s │ │ + │ └────────────────────────┘ │ + └─────────────────────────────┘ +``` + +- **Bugger** is the SSE *client* (`Sources/Services/BitableEventService.swift`, 227 LOC). +- **bugger-feishu** is the SSE *server* — a separate FastAPI service deployed as a k8s pod. +- The ingress sits between them and **will close any connection that appears idle** for + longer than its configured idle-timeout (`proxy_read_timeout` for nginx ingress, default + 60s; cloud LBs vary). + +--- + +## 2. Heartbeat Analysis + +### 2.1 Client side — relies on server, sends nothing + +SSE is a one-way protocol (server → client). Bugger's client **cannot keep the connection +alive from its end** — it can only receive. It depends entirely on the server periodically +sending bytes to reset the ingress's idle timer. + +Client-side idle-timeout configuration (`BitableEventService.swift:61-66`): +```swift +let config = URLSessionConfiguration.ephemeral +// The server sends a heartbeat every ~30s, so 5 min of silence is a +// safe "something went wrong" threshold. +config.timeoutIntervalForRequest = 300 +config.timeoutIntervalForResource = .infinity +config.waitsForConnectivity = true +``` +- `timeoutIntervalForRequest = 300` (5 min): the client's own dead-connection detector. + Set generously above the 30s server heartbeat so it never fires under normal operation. +- `timeoutIntervalForResource = .infinity`: unbounded resource lifetime (correct for a stream). +- `waitsForConnectivity = true`: OS waits through transient network gaps rather than failing. + +The client's 300s timeout is **never the bottleneck** — it is the ingress that will drop +the connection first. + +### 2.2 Server side — specified, must be verified deployed + +The heartbeat is specified in `docs/BITABLE_CHANGE_NOTIFICATION_IMPL.md:686-693`: +```python +# Wait for events with heartbeat timeout +event = await asyncio.wait_for(conn.queue.get(), timeout=30.0) +yield f"event: {event['event']}\ndata: {event['data']}\n\n" +except asyncio.TimeoutError: + # Heartbeat + yield ": heartbeat\n\n" +``` +- **Interval: 30 seconds.** Emits an SSE comment frame (`: heartbeat\n\n`) when no real + event is queued within 30s. +- Response headers include `Connection: keep-alive` and `X-Accel-Buffering: no` + (lines 707-711) to prevent buffering proxies from holding the heartbeat. + +⚠️ **This code lives in the separate bugger-feishu repo and is only *specified* in this +repo's docs.** Whether it is actually deployed must be verified in the bugger-feishu repo +before relying on it. + +### 2.3 Will the heartbeat prevent ingress disconnect? + +The ingress closes an upstream connection when **no bytes flow** for its idle-timeout +period. The heartbeat resets that timer each time it is sent. + +| Ingress idle-timeout | Heartbeat (30s) prevents disconnect? | +|---|---| +| ≥ 60s (nginx ingress default) | ✅ Yes — 30s < 60s, timer resets each cycle | +| 30s | ⚠️ Marginal — race conditions; not safe | +| < 30s (some cloud LBs) | ❌ No — connection dropped every cycle | + +**Requirements for the heartbeat to work:** +1. Ingress `proxy_read_timeout` / LB idle-timeout **≥ 60s** (30s heartbeat + safety margin). +2. `proxy_buffering off` on the ingress location, **or** the server's `X-Accel-Buffering: no` + header must be honored — otherwise heartbeats are buffered and never reach the client + until the response completes (which never happens for a stream). +3. The 30s heartbeat must actually be running in the deployed bugger-feishu pod. + +### 2.4 What happens on a too-tight ingress timeout? + +If the ingress idle-timeout is, say, 15s: +1. Bugger connects. +2. After 15s of silence (before the 30s heartbeat fires), the ingress closes the upstream. +3. The client sees the stream end → schedules a reconnect in 30s. +4. Repeat every ~45s indefinitely — a steady-state reconnect loop. + +This wastes bandwidth, defeats the push model, and — with many Bugger clients — creates a +synchronized reconnect storm (no jitter, see §3.3). + +--- + +## 3. Reconnection Analysis + +### 3.1 Mechanism — present and well-guarded + +Reconnection is implemented in `BitableEventService.swift`. On any stream end (error or +clean), if the user still intends to be connected, a reconnect is scheduled. + +State guard (`BitableEventService.swift:21-23`): +```swift +/// The user's intent to be connected. Distinguishes an unexpected stream +/// drop (should auto-reconnect) from an explicit `disconnect()` (should not). +private var isEnabled = false +``` + +Reconnect scheduling (`BitableEventService.swift:102-124`): +```swift +private func handleStreamEnd(error: Error?) { + task = nil + session = nil + delegate = nil + + if let error { + BuggerLog.error("BitableEventService: stream ended (\(error.localizedDescription))") + } else { + BuggerLog.info("BitableEventService: stream ended") + } + + // Only auto-reconnect if the user still wants to be connected. + guard isEnabled else { return } + + let work = DispatchWorkItem { [weak self] in + guard let self, self.isEnabled else { return } + guard let url = self.buildURL() else { return } + BuggerLog.info("BitableEventService: reconnecting...") + self.openStream(at: url) + } + reconnectWorkItem = work + DispatchQueue.main.asyncAfter(deadline: .now() + reconnectDelay, execute: work) +} +``` + +Fixed delay constant (`BitableEventService.swift:25`): +```swift +private let reconnectDelay: TimeInterval = 30 +``` + +**Guard quality — good:** +- Double-guarded against intentional disconnect: `isEnabled` check at line 114, plus + re-checked inside the `DispatchWorkItem` at line 117. +- Pending reconnect is cancellable: `closeStream()` calls `reconnectWorkItem?.cancel()` + (line 93), so `disconnect()` cannot race with a scheduled reconnect. +- All state mutation happens on the main queue (callbacks routed via `DispatchQueue.main.async` + in `SSESessionDelegate` at lines 202, 223). + +**Gap — no catch-up fetch:** reconnection does **not** trigger an immediate `fetchNow()` +to recover events missed during the downtime window. It relies on the next server push or +the periodic `PollerService` to eventually resync. This is acceptable given polling is the +fallback, but means event freshness is bounded by the poll interval after any reconnect. + +### 3.2 Exponential backoff — NOT implemented + +The reconnect delay is a **constant 30s, every time, forever**, regardless of how many +consecutive failures occur. + +Searching the repo, "exponential backoff" appears only in design docs for the *rejected* +direct-WebSocket design — never in shipped SSE code: +- `docs/TECH_INVESTIGATION.md:56-58` — Feishu raw WebSocket option (not chosen) +- `docs/TECH_INVESTIGATION.md:424` — planned task for the abandoned WSClient approach +- `docs/TECH_INVESTIGATION.md:544` — same + +### 3.3 Risks of the fixed-delay strategy + +| Scenario | Behavior | Problem | +|---|---|---| +| bugger-feishu pod down 1 hour | 120 reconnect attempts at fixed 30s | No de-escalation; wastes resources | +| bugger-feishu pod restarts | All clients reconnect simultaneously after 30s | **Thundering herd** — no jitter | +| Ingress drops every cycle (timeout < 30s) | Reconnect every ~45s per client | Steady-state storm across all clients | +| Many clients, transient outage | All retry on identical 30s cadence | Synchronized load spikes on ingress | + +The lack of jitter is the most operationally significant gap: with N Bugger clients all +using the same fixed 30s delay, a pod restart causes all N to hit the ingress at the same +instant, every 30s, until one succeeds. + +--- + +## 4. Findings Summary + +### Q1 — Heartbeat & ingress timeout + +- **Yes**, a heartbeat exists — but it is **server-side only** (bugger-feishu emits + `: heartbeat\n\n` every ~30s). Bugger's client sends nothing. +- **Will it prevent ingress disconnect?** Only if the ingress idle-timeout + (`proxy_read_timeout` / LB idle-timeout) is **≥ the heartbeat interval (30s)**, with + margin — recommend **≥ 60s**. It also requires `proxy_buffering off` (or honored + `X-Accel-Buffering: no`) so heartbeats are not buffered. +- **Caveat:** the heartbeat code lives in the separate bugger-feishu repo. It is only + *specified* in this repo's docs — verify it is actually deployed. + +### Q2 — Reconnection & exponential backoff + +- **Yes**, reconnection exists. It is triggered on stream end when `isEnabled == true`, + uses a fixed **30s delay**, and is properly guarded against intentional disconnects + (double guard + cancellable work item). +- **No**, it does **not** use exponential backoff. The delay is constant 30s, indefinitely, + with no cap, no jitter, and no escalation/de-escalation. This creates retry-storm and + thundering-herd risks under sustained or synchronized outages. + +--- + +## 5. Recommendations + +### R1 — Ingress / LB configuration (ops, no code change) + +Configure the ingress in front of bugger-feishu: +1. `proxy_read_timeout 60s;` (or higher) — must exceed the 30s heartbeat with margin. +2. `proxy_buffering off;` on the SSE location — or ensure `X-Accel-Buffering: no` is honored. +3. `proxy_send_timeout 60s;` — symmetric. +4. Verify the cloud LB idle-timeout (if any, in front of the ingress) is also ≥ 60s — some + LBs default to 30s or less and override the ingress setting. + +### R2 — Verify heartbeat is deployed (ops, no code change) + +Confirm in the bugger-feishu repo that the 30s `: heartbeat\n\n` emission is actually +running in the deployed image, not just specified in this repo's docs. If absent, the +ingress will drop the connection on every idle cycle regardless of R1. + +### R3 — Add exponential backoff with jitter and cap (code change) + +Modify `BitableEventService.swift` to replace the fixed `reconnectDelay` with a backoff +strategy. Suggested parameters: + +``` +attempt 1: delay = 1s (±20% jitter) +attempt 2: delay = 2s (±20% jitter) +attempt 3: delay = 4s (±20% jitter) +attempt 4: delay = 8s (±20% jitter) +attempt 5: delay = 16s (±20% jitter) +attempt 6+: delay = 30s (±20% jitter) ← cap +``` + +- Base: `delay = min(cap, base * 2^(attempt-1))` +- Cap: 30s (preserves current steady-state behavior) +- Jitter: ±20% of computed delay (prevents thundering herd) +- Reset attempt counter on a successful `connected` event from the server. + +### R4 — Catch-up fetch on reconnect (optional code change) + +On a successful reconnect, trigger `PollerService.shared.fetchNow()` once to recover any +events missed during the downtime window, rather than waiting for the next poll cycle. +This bounds event freshness loss to one fetch rather than the full poll interval. + +--- + +## 6. Key Code Locations + +All paths in `/Users/tigeren/Dev/xorbitlab/bugger/`: + +| Location | What | +|---|---| +| `Sources/Services/BitableEventService.swift:25` | `reconnectDelay = 30` (fixed, no backoff) | +| `Sources/Services/BitableEventService.swift:21-23` | `isEnabled` guard | +| `Sources/Services/BitableEventService.swift:61-66` | Client idle-timeout config (300s request, infinite resource) | +| `Sources/Services/BitableEventService.swift:83` | Request timeout (300s) | +| `Sources/Services/BitableEventService.swift:92-100` | `closeStream` — cancels pending reconnect | +| `Sources/Services/BitableEventService.swift:102-124` | `handleStreamEnd` — auto-reconnect scheduling | +| `Sources/Services/BitableEventService.swift:126-138` | `handleEvent` — `change`→`fetchNow`, heartbeat no-op | +| `Sources/Services/BitableEventService.swift:170-226` | `SSESessionDelegate` — frame parser, comment/heartbeat ignored | +| `Sources/AppDelegate.swift:33` | `connect()` on app launch | +| `Sources/Views/Settings/SettingsView.swift:293` | `reconnect()` on settings save | +| `Sources/Views/Settings/SettingsView.swift:181` | `disconnect()` on field clear | +| `docs/BITABLE_CHANGE_NOTIFICATION_IMPL.md:686-693` | Server heartbeat spec (30s, must verify deployed) | +| `docs/BITABLE_CHANGE_NOTIFICATION_IMPL.md:707-711` | Server SSE response headers (keep-alive, no buffering) | +| `docs/TECH_INVESTIGATION.md:56-58,424,544` | Backoff references (rejected WSClient design only) |