diff --git a/Pulso.xcodeproj/project.pbxproj b/Pulso.xcodeproj/project.pbxproj index 09add4b..56ae45d 100644 --- a/Pulso.xcodeproj/project.pbxproj +++ b/Pulso.xcodeproj/project.pbxproj @@ -43,6 +43,14 @@ B20000000000000000000005 /* ObjC.swift in Sources */ = {isa = PBXBuildFile; fileRef = F20000000000000000000005 /* ObjC.swift */; }; B20000000000000000000006 /* ObjCExceptions.m in Sources */ = {isa = PBXBuildFile; fileRef = F20000000000000000000006 /* ObjCExceptions.m */; }; B20000000000000000000007 /* HealthKitCatalogTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = F20000000000000000000009 /* HealthKitCatalogTests.swift */; }; + B20000000000000000000008 /* Destination.swift in Sources */ = {isa = PBXBuildFile; fileRef = F2000000000000000000000A /* Destination.swift */; }; + B20000000000000000000009 /* Transports.swift in Sources */ = {isa = PBXBuildFile; fileRef = F2000000000000000000000B /* Transports.swift */; }; + B2000000000000000000000A /* MQTTPublisher.swift in Sources */ = {isa = PBXBuildFile; fileRef = F2000000000000000000000C /* MQTTPublisher.swift */; }; + B2000000000000000000000B /* OutboxHub.swift in Sources */ = {isa = PBXBuildFile; fileRef = F2000000000000000000000D /* OutboxHub.swift */; }; + B2000000000000000000000C /* DestinationsView.swift in Sources */ = {isa = PBXBuildFile; fileRef = F2000000000000000000000E /* DestinationsView.swift */; }; + B2000000000000000000000D /* MQTTPacketTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = F2000000000000000000000F /* MQTTPacketTests.swift */; }; + B2000000000000000000000E /* DestinationTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = F20000000000000000000010 /* DestinationTests.swift */; }; + B2000000000000000000000F /* Intents.swift in Sources */ = {isa = PBXBuildFile; fileRef = F20000000000000000000011 /* Intents.swift */; }; /* End PBXBuildFile section */ /* Begin PBXContainerItemProxy section */ @@ -98,6 +106,14 @@ F20000000000000000000007 /* ObjCExceptions.h */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.c.h; path = ObjCExceptions.h; sourceTree = ""; }; F20000000000000000000008 /* Pulso-Bridging-Header.h */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.c.h; path = Pulso-Bridging-Header.h; sourceTree = ""; }; F20000000000000000000009 /* HealthKitCatalogTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = HealthKitCatalogTests.swift; sourceTree = ""; }; + F2000000000000000000000A /* Destination.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = Destination.swift; sourceTree = ""; }; + F2000000000000000000000B /* Transports.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = Transports.swift; sourceTree = ""; }; + F2000000000000000000000C /* MQTTPublisher.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = MQTTPublisher.swift; sourceTree = ""; }; + F2000000000000000000000D /* OutboxHub.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = OutboxHub.swift; sourceTree = ""; }; + F2000000000000000000000E /* DestinationsView.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = DestinationsView.swift; sourceTree = ""; }; + F2000000000000000000000F /* MQTTPacketTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = MQTTPacketTests.swift; sourceTree = ""; }; + F20000000000000000000010 /* DestinationTests.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = DestinationTests.swift; sourceTree = ""; }; + F20000000000000000000011 /* Intents.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = Intents.swift; sourceTree = ""; }; /* End PBXFileReference section */ /* Begin PBXFrameworksBuildPhase section */ @@ -135,6 +151,7 @@ A10000000000000000000004 /* Model */, A10000000000000000000005 /* Health */, A10000000000000000000006 /* Sync */, + A2000000000000000000000B /* Destinations */, A10000000000000000000007 /* Support */, A10000000000000000000008 /* UI */, F10000000000000000000017 /* Assets.xcassets */, @@ -147,6 +164,7 @@ A10000000000000000000003 /* App */ = { isa = PBXGroup; children = ( + F20000000000000000000011 /* Intents.swift */, F10000000000000000000001 /* PulsoApp.swift */, F10000000000000000000002 /* AppServices.swift */, ); @@ -182,6 +200,7 @@ A10000000000000000000006 /* Sync */ = { isa = PBXGroup; children = ( + F2000000000000000000000D /* OutboxHub.swift */, F1000000000000000000000B /* SyncEngine.swift */, F1000000000000000000000C /* Outbox.swift */, F1000000000000000000000D /* Uploader.swift */, @@ -207,6 +226,7 @@ A10000000000000000000008 /* UI */ = { isa = PBXGroup; children = ( + F2000000000000000000000E /* DestinationsView.swift */, F10000000000000000000013 /* RootView.swift */, F10000000000000000000014 /* StatusView.swift */, F10000000000000000000015 /* SettingsView.swift */, @@ -218,6 +238,8 @@ A10000000000000000000009 /* PulsoTests */ = { isa = PBXGroup; children = ( + F20000000000000000000010 /* DestinationTests.swift */, + F2000000000000000000000F /* MQTTPacketTests.swift */, F20000000000000000000009 /* HealthKitCatalogTests.swift */, F1000000000000000000001C /* SerializerTests.swift */, F1000000000000000000001D /* OutboxTests.swift */, @@ -238,6 +260,16 @@ name = Products; sourceTree = ""; }; + A2000000000000000000000B /* Destinations */ = { + isa = PBXGroup; + children = ( + F2000000000000000000000C /* MQTTPublisher.swift */, + F2000000000000000000000B /* Transports.swift */, + F2000000000000000000000A /* Destination.swift */, + ); + path = Destinations; + sourceTree = ""; + }; /* End PBXGroup section */ /* Begin PBXNativeTarget section */ @@ -329,6 +361,12 @@ isa = PBXSourcesBuildPhase; buildActionMask = 2147483647; files = ( + B2000000000000000000000F /* Intents.swift in Sources */, + B2000000000000000000000C /* DestinationsView.swift in Sources */, + B2000000000000000000000B /* OutboxHub.swift in Sources */, + B2000000000000000000000A /* MQTTPublisher.swift in Sources */, + B20000000000000000000009 /* Transports.swift in Sources */, + B20000000000000000000008 /* Destination.swift in Sources */, B20000000000000000000006 /* ObjCExceptions.m in Sources */, B20000000000000000000005 /* ObjC.swift in Sources */, B20000000000000000000004 /* HealthAuthorization.swift in Sources */, @@ -364,6 +402,8 @@ isa = PBXSourcesBuildPhase; buildActionMask = 2147483647; files = ( + B2000000000000000000000E /* DestinationTests.swift in Sources */, + B2000000000000000000000D /* MQTTPacketTests.swift in Sources */, B20000000000000000000007 /* HealthKitCatalogTests.swift in Sources */, B1000000000000000000001C /* SerializerTests.swift in Sources */, B1000000000000000000001D /* OutboxTests.swift in Sources */, diff --git a/Pulso/App/AppServices.swift b/Pulso/App/AppServices.swift index f4784e6..d9b529f 100644 --- a/Pulso/App/AppServices.swift +++ b/Pulso/App/AppServices.swift @@ -5,12 +5,16 @@ import SwiftUI /// Composition root: owns the four components and the lifecycle glue. @MainActor final class AppServices { + /// One instance per process, shared by the app delegate and App + /// Intents (which can run with no UI shown). + static let shared = AppServices() + let healthStore = HKHealthStore() let settings = AppSettings() let status = AppStatus() let log: LogStore let anchors: AnchorStore - let outbox: Outbox + let outbox: OutboxHub let engine: SyncEngine let hub: TriggerHub private let uploader = HTTPUploader() @@ -29,25 +33,39 @@ final class AppServices { log = LogStore(directory: root) anchors = AnchorStore(directory: root.appendingPathComponent("Anchors", isDirectory: true)) - outbox = Outbox( - directory: root.appendingPathComponent("Outbox", isDirectory: true), - anchors: anchors, - uploader: uploader, - target: { AppSettings.currentTarget() }, - log: log, - status: status - ) + let queues = root.appendingPathComponent("Outboxes", isDirectory: true) + Self.adoptLegacyOutbox(from: root.appendingPathComponent("Outbox", isDirectory: true), into: queues) + outbox = OutboxHub(root: queues, log: log, status: status) let uploader = uploader engine = SyncEngine( healthStore: healthStore, anchors: anchors, outbox: outbox, log: log, status: status, - latestHints: { - guard let base = AppSettings.currentBase() else { return nil } - return await uploader.fetchLatest(base: base.url, token: base.token) + latestHints: { destinations in + // The anchor is shared, so skipping history is only safe when + // every destination already has it: all must answer, and the + // earliest answer per type wins. + guard !destinations.isEmpty, destinations.allSatisfy({ $0.latestURL != nil }) else { return nil } + var answers: [[String: String]] = [] + for destination in destinations { + guard let url = destination.latestURL, + let hints = await uploader.fetchLatest(url: url, token: destination.bearerToken) else { return nil } + answers.append(hints) + } + return CatchUp.earliest(answers) } ) hub = TriggerHub(healthStore: healthStore, engine: engine, log: log) status.load() + let device = UIDevice.current + let info = Bundle.main.infoDictionary ?? [:] + DeviceInfo.current = DeviceInfo( + id: device.identifierForVendor?.uuidString ?? "unknown", + name: device.name, + model: device.model, + systemVersion: device.systemVersion, + appVersion: "\(info["CFBundleShortVersionString"] as? String ?? "?") (\(info["CFBundleVersion"] as? String ?? "?"))", + bundleId: Bundle.main.bundleIdentifier ?? "" + ) let status = status let log = log Task { @@ -59,20 +77,53 @@ final class AppServices { } } + /// Batches queued by the single-server version belong to the migrated + /// "Pulso server" destination. + private static func adoptLegacyOutbox(from legacy: URL, into queues: URL) { + let fm = FileManager.default + guard let files = try? fm.contentsOfDirectory(at: legacy, includingPropertiesForKeys: nil) else { return } + // Without a migrated server there is nowhere to deliver them yet: + // leave the folder alone rather than lose queued data. + guard let destination = AppSettings.currentDestinations().first(where: { $0.kind == .pulso }) else { return } + // Never move a queue's sequence backwards, or new batches would sort + // ahead of ones already waiting. + let defaults = UserDefaults.standard + let sequenceKey = "\(SettingsKeys.outboxSequence).\(destination.id.uuidString)" + defaults.set( + max(defaults.integer(forKey: SettingsKeys.outboxSequence), defaults.integer(forKey: sequenceKey)), + forKey: sequenceKey + ) + do { + let target = queues.appendingPathComponent(destination.id.uuidString, isDirectory: true) + try fm.createDirectory(at: target, withIntermediateDirectories: true) + for file in files where file.pathExtension == "json" { + try fm.moveItem(at: file, to: target.appendingPathComponent(file.lastPathComponent)) + } + } catch { + return // whatever didn't move stays put; the next launch tries again + } + try? fm.removeItem(at: legacy) + } + private var isTestRun: Bool { ProcessInfo.processInfo.environment["XCTestConfigurationFilePath"] != nil } + private var launched = false + /// Called from application(_:didFinishLaunchingWithOptions:). Also runs /// when HealthKit relaunches the app in the background for delivery. func launch() { + guard !launched else { return } + launched = true guard HKHealthStore.isHealthDataAvailable() else { status.healthUnavailable = true return } guard !isTestRun else { return } - hub.registerBackgroundRefresh() + hub.registerBackgroundTasks() hub.scheduleNextRefresh() + hub.scheduleProcessing() Task { await refreshAuthStatus() await hub.startObservers() @@ -90,6 +141,7 @@ final class AppServices { } case .background: hub.scheduleNextRefresh() + hub.scheduleProcessing() default: break } @@ -188,10 +240,122 @@ final class AppServices { await syncNow(.reimport) } - func testConnection() async -> String { - guard let url = settings.healthURL else { - return "Enter a valid server URL first" + // MARK: - Shortcuts + + /// Intents may run before the delegate did (or in a process HealthKit + /// woke); launching twice is a no-op. + func launchIfNeeded() { + launch() + } + + func syncFromIntent(keys: Set?) async -> String { + // Otherwise the pass returns early and the shortcut would report the + // previous sync's summary as if this one had run. + guard await outbox.hasActiveDestination() else { return "No destination is configured." } + await withBackgroundTime("pulso-shortcut-sync") { + await engine.requestSync(.intent("Sync Health Data"), only: keys) + } + await refreshDerived() + return status.lastSyncSummary ?? "Sync finished." + } + + func resend(keys: Set?, start: Date?, end: Date?, reason: SyncReason) async -> String { + var summary = "" + await withBackgroundTime("pulso-shortcut-resend") { + summary = await engine.resend(only: keys, start: start, end: end, reason: reason) + } + await refreshDerived() + return summary + } + + func statusReport() async -> String { + await outbox.refreshPendingStatus() + var lines: [String] = [] + if let finished = status.lastSyncFinished { + lines.append("Last sync finished \(finished.formatted(date: .abbreviated, time: .shortened)).") + } else { + lines.append("No sync has finished yet.") + } + if let summary = status.lastSyncSummary { lines.append(summary) } + for destination in settings.destinations where destination.enabled { + let state = status.destinationStatus[destination.id] ?? DestinationStatus() + var line = "\(destination.name): \(state.pending) batch(es) queued" + if let success = state.lastSuccess { + line += ", last delivery \(success.formatted(date: .abbreviated, time: .shortened))" + } + if let failure = state.lastFailure, failure > (state.lastSuccess ?? .distantPast), let response = state.lastResponse { + line += ", failing: \(response)" + } + lines.append(line + ".") + } + if settings.destinations.allSatisfy({ !$0.enabled }) { + lines.append("No destination is enabled.") + } + return lines.joined(separator: "\n") + } + + /// Keeps the process alive while a shortcut's work runs in the background. + private func withBackgroundTime(_ name: String, _ work: () async -> Void) async { + // Ended either when the work finishes or when iOS says time is up, + // whichever comes first; an outstanding assertion gets the app killed. + final class Assertion: @unchecked Sendable { + private let lock = NSLock() + private var id: UIBackgroundTaskIdentifier = .invalid + func set(_ value: UIBackgroundTaskIdentifier) { lock.lock(); id = value; lock.unlock() } + func end() { + lock.lock() + let current = id + id = .invalid + lock.unlock() + if current != .invalid { UIApplication.shared.endBackgroundTask(current) } + } + } + let assertion = Assertion() + assertion.set(UIApplication.shared.beginBackgroundTask(withName: name) { assertion.end() }) + await work() + assertion.end() + } + + // MARK: - Destinations + + func destinationsChanged() { + Task { + await outbox.refreshPendingStatus() + } + } + + func removeDestination(_ destination: Destination) { + settings.destinations.removeAll { $0.id == destination.id } + destination.deleteSecrets() + Task { await outbox.forget(destination.id) } + } + + /// Sends something harmless the destination must accept: an empty test + /// batch over HTTP (signed and authenticated like real ones), a message + /// on /test over MQTT, a small file for file export. + func testDestination(_ destination: Destination) async -> String { + do { + switch destination.kind { + case .pulso: + guard let base = AppSettings.baseURL(from: destination.url), + let url = URL(string: base.absoluteString + "/health") else { return "Enter a valid URL first" } + return await uploader.checkHealth(url) + case .rest, .homeAssistant: + let transport = HTTPTransport(destination: destination, session: HTTPTransport.makeSession()) + let format: Destination.BodyFormat = destination.kind == .homeAssistant ? .envelope : destination.format + let result = try await transport.post(try BatchBody.test(device: DeviceInfo.current, format: format)) + return result.isSuccess + ? "OK — HTTP \(result.status)" + : "HTTP \(result.status): \(result.bodySnippet.prefix(120))" + case .mqtt: + try await MQTTTransport(destination: destination).test() + return "OK — published to \(MQTTTransport(destination: destination).topic("test"))" + case .file: + try FileTransport(destination: destination).test() + return "OK — wrote pulso-test.txt" + } + } catch { + return "Failed: \(error.localizedDescription)" } - return await uploader.checkHealth(url) } } diff --git a/Pulso/App/Intents.swift b/Pulso/App/Intents.swift new file mode 100644 index 0000000..aa280e8 --- /dev/null +++ b/Pulso/App/Intents.swift @@ -0,0 +1,142 @@ +import AppIntents +import Foundation + +/// A syncable type as Shortcuts sees it (sample types and snapshots). +struct HealthTypeEntity: AppEntity, Identifiable { + static var typeDisplayRepresentation: TypeDisplayRepresentation = "Health Data Type" + static var defaultQuery = HealthTypeQuery() + + var id: String + var name: String + var group: String + + var displayRepresentation: DisplayRepresentation { + DisplayRepresentation(title: "\(name)", subtitle: "\(group) · \(id)") + } + + static var all: [HealthTypeEntity] { + TypeRegistry.snapshots.map { HealthTypeEntity(id: $0.key, name: $0.displayName, group: $0.group) } + + TypeRegistry.all.map { HealthTypeEntity(id: $0.key, name: $0.displayName, group: $0.group) } + } +} + +struct HealthTypeQuery: EntityStringQuery { + func entities(for identifiers: [String]) async throws -> [HealthTypeEntity] { + let wanted = Set(identifiers) + return HealthTypeEntity.all.filter { wanted.contains($0.id) } + } + + func entities(matching string: String) async throws -> [HealthTypeEntity] { + HealthTypeEntity.all.filter { + $0.name.localizedCaseInsensitiveContains(string) || $0.id.localizedCaseInsensitiveContains(string) + } + } + + func suggestedEntities() async throws -> [HealthTypeEntity] { + HealthTypeEntity.all + } +} + +/// Incremental sync, optionally limited to some types. With a date range it +/// re-sends what HealthKit holds in that range instead (anchors untouched). +struct SyncHealthDataIntent: AppIntent { + static var title: LocalizedStringResource = "Sync Health Data" + static var description = IntentDescription( + "Sends new Apple Health data to your destinations. Pick types to limit it; add a date range to re-send that period." + ) + static var openAppWhenRun = false + + @Parameter(title: "Types", description: "Leave empty for every enabled type.") + var types: [HealthTypeEntity]? + + @Parameter(title: "From") + var start: Date? + + @Parameter(title: "Until") + var end: Date? + + static var parameterSummary: some ParameterSummary { + Summary("Sync \(\.$types) from \(\.$start) until \(\.$end)") + } + + @MainActor + func perform() async throws -> some IntentResult & ReturnsValue & ProvidesDialog { + let services = AppServices.shared + services.launchIfNeeded() + let keys = types.map { Set($0.map(\.id)) } + let summary: String + if start != nil || end != nil { + summary = await services.resend(keys: keys, start: start, end: end, reason: .intent("Sync Health Data")) + } else { + summary = await services.syncFromIntent(keys: keys) + } + return .result(value: summary, dialog: IntentDialog(stringLiteral: summary)) + } +} + +/// Re-sends history (all of it by default) for some or all types. Anchors +/// are left alone; receivers dedupe by uuid. +struct BackfillHealthDataIntent: AppIntent { + static var title: LocalizedStringResource = "Backfill Health Data" + static var description = IntentDescription( + "Re-sends Apple Health history to your destinations. Large histories take a while: keep the phone unlocked." + ) + /// A full backfill outlives background time; running in the foreground + /// keeps HealthKit readable and the app alive. + static var openAppWhenRun = true + + @Parameter(title: "Types", description: "Leave empty for every enabled type.") + var types: [HealthTypeEntity]? + + @Parameter(title: "Since", description: "Leave empty for all history.") + var start: Date? + + static var parameterSummary: some ParameterSummary { + Summary("Backfill \(\.$types) since \(\.$start)") + } + + @MainActor + func perform() async throws -> some IntentResult & ReturnsValue & ProvidesDialog { + let services = AppServices.shared + services.launchIfNeeded() + let summary = await services.resend( + keys: types.map { Set($0.map(\.id)) }, start: start, end: nil, reason: .intent("Backfill") + ) + return .result(value: summary, dialog: IntentDialog(stringLiteral: summary)) + } +} + +struct GetSyncStatusIntent: AppIntent { + static var title: LocalizedStringResource = "Get Last Sync Status" + static var description = IntentDescription("Reports when Pulso last synced, what it sent, and what is still queued.") + static var openAppWhenRun = false + + @MainActor + func perform() async throws -> some IntentResult & ReturnsValue & ProvidesDialog { + let summary = await AppServices.shared.statusReport() + return .result(value: summary, dialog: IntentDialog(stringLiteral: summary)) + } +} + +struct PulsoShortcuts: AppShortcutsProvider { + static var appShortcuts: [AppShortcut] { + AppShortcut( + intent: SyncHealthDataIntent(), + phrases: ["Sync health data with \(.applicationName)", "\(.applicationName) sync"], + shortTitle: "Sync Health Data", + systemImageName: "arrow.triangle.2.circlepath" + ) + AppShortcut( + intent: BackfillHealthDataIntent(), + phrases: ["Backfill health data with \(.applicationName)"], + shortTitle: "Backfill", + systemImageName: "clock.arrow.circlepath" + ) + AppShortcut( + intent: GetSyncStatusIntent(), + phrases: ["\(.applicationName) sync status"], + shortTitle: "Sync Status", + systemImageName: "checkmark.circle" + ) + } +} diff --git a/Pulso/App/PulsoApp.swift b/Pulso/App/PulsoApp.swift index c6456b2..e313e5d 100644 --- a/Pulso/App/PulsoApp.swift +++ b/Pulso/App/PulsoApp.swift @@ -17,7 +17,7 @@ struct PulsoApp: App { @MainActor final class AppDelegate: NSObject, UIApplicationDelegate { - let services = AppServices() + let services = AppServices.shared func application( _ application: UIApplication, diff --git a/Pulso/Destinations/Destination.swift b/Pulso/Destinations/Destination.swift new file mode 100644 index 0000000..bebce8e --- /dev/null +++ b/Pulso/Destinations/Destination.swift @@ -0,0 +1,229 @@ +import Foundation + +/// One place health data goes. Several can be enabled at once; each has +/// its own durable queue, so a slow or broken one never holds up the rest. +/// Secrets (bearer token, HMAC secret, MQTT password) live in the Keychain +/// under the destination's id, never in this struct. +struct Destination: Codable, Identifiable, Equatable, Sendable { + enum Kind: String, Codable, CaseIterable, Sendable { + /// Any HTTPS endpoint: POST of the pulso.v2 envelope (or a bare array). + case rest + /// A server speaking the original Pulso protocol (POST /ingest). + case pulso + /// A Home Assistant webhook trigger. + case homeAssistant + /// An MQTT broker, one message per type per batch. + case mqtt + /// NDJSON files in a folder (Files app, iCloud Drive). + case file + + var title: String { + switch self { + case .rest: "HTTPS / webhook" + case .pulso: "Pulso server" + case .homeAssistant: "Home Assistant webhook" + case .mqtt: "MQTT" + case .file: "File export (NDJSON)" + } + } + } + + enum BodyFormat: String, Codable, CaseIterable, Sendable { + /// {"schema":"pulso.v2","device":{...},"type":...,"items":[...]} + case envelope + /// The original Pulso body: a bare JSON array of items. + case array + /// One JSON object per line. + case ndjson + } + + /// Header values can be API keys, so they are kept in the Keychain: + /// AppSettings stores them there and saves the list with values blank. + struct Header: Codable, Equatable, Hashable, Identifiable, Sendable { + var id = UUID() + var name: String + var value: String + + init(name: String, value: String) { + self.name = name + self.value = value + } + + init(from decoder: Decoder) throws { + let c = try decoder.container(keyedBy: CodingKeys.self) + id = try c.decodeIfPresent(UUID.self, forKey: .id) ?? UUID() + name = try c.decodeIfPresent(String.self, forKey: .name) ?? "" + value = try c.decodeIfPresent(String.self, forKey: .value) ?? "" + } + } + + var id = UUID() + var name: String + var kind: Kind + var enabled = true + /// Set once the destination has had a usable URL / host. Only then does + /// it keep queueing while unconfigured (e.g. mid-edit); one added and + /// never set up gets nothing, so its queue can't grow forever. + var wasConfigured = false + + // HTTP kinds + var url = "" + var format: BodyFormat = .envelope + var gzip = true + var headers: [Header] = [] + /// Ask /latest what the server already has when a type has no + /// anchor yet (fresh install), instead of re-sending all history. + var catchUp = false + + // MQTT + var mqttHost = "" + var mqttPort = 8883 + var mqttTLS = true + var mqttUsername = "" + var mqttTopicPrefix = "pulso" + var mqttRetain = false + + // File export: a security-scoped bookmark of a folder the user picked; + // nil means the app's own Documents/Pulso Export folder. + var folderBookmark: Data? + var folderName: String? + + /// Every field falls back to its default when missing, so a field added + /// in a later version can't make an existing list fail to decode. + init(from decoder: Decoder) throws { + let c = try decoder.container(keyedBy: CodingKeys.self) + let kind = try c.decode(Kind.self, forKey: .kind) + self.init(name: try c.decodeIfPresent(String.self, forKey: .name) ?? kind.title, kind: kind) + id = try c.decode(UUID.self, forKey: .id) + enabled = try c.decodeIfPresent(Bool.self, forKey: .enabled) ?? enabled + url = try c.decodeIfPresent(String.self, forKey: .url) ?? url + format = (try? c.decodeIfPresent(BodyFormat.self, forKey: .format)) ?? format + gzip = try c.decodeIfPresent(Bool.self, forKey: .gzip) ?? gzip + headers = try c.decodeIfPresent([Header].self, forKey: .headers) ?? headers + catchUp = try c.decodeIfPresent(Bool.self, forKey: .catchUp) ?? catchUp + mqttHost = try c.decodeIfPresent(String.self, forKey: .mqttHost) ?? mqttHost + mqttPort = try c.decodeIfPresent(Int.self, forKey: .mqttPort) ?? mqttPort + mqttTLS = try c.decodeIfPresent(Bool.self, forKey: .mqttTLS) ?? mqttTLS + mqttUsername = try c.decodeIfPresent(String.self, forKey: .mqttUsername) ?? mqttUsername + mqttTopicPrefix = try c.decodeIfPresent(String.self, forKey: .mqttTopicPrefix) ?? mqttTopicPrefix + mqttRetain = try c.decodeIfPresent(Bool.self, forKey: .mqttRetain) ?? mqttRetain + folderBookmark = try c.decodeIfPresent(Data.self, forKey: .folderBookmark) + folderName = try c.decodeIfPresent(String.self, forKey: .folderName) + // Last: the fallback reads the fields above. + wasConfigured = try c.decodeIfPresent(Bool.self, forKey: .wasConfigured) ?? isConfigured + } + + init(name: String, kind: Kind) { + self.name = name + self.kind = kind + switch kind { + case .pulso: + format = .array + catchUp = true + case .homeAssistant: + gzip = false + case .file: + format = .ndjson + case .rest, .mqtt: + break + } + } + + /// Items per uploaded batch. Home Assistant automations choke on huge + /// bodies; everything else takes the outbox's normal batches. + var maxItemsPerBatch: Int? { + kind == .homeAssistant ? 200 : nil + } + + // MARK: - Secrets + + private func account(_ name: String) -> String { "destination.\(id.uuidString).\(name)" } + + var bearerToken: String? { + get { Keychain.string(for: account("bearer")) } + nonmutating set { Keychain.set(newValue, for: account("bearer")) } + } + + var hmacSecret: String? { + get { Keychain.string(for: account("hmac")) } + nonmutating set { Keychain.set(newValue, for: account("hmac")) } + } + + var mqttPassword: String? { + get { Keychain.string(for: account("mqtt")) } + nonmutating set { Keychain.set(newValue, for: account("mqtt")) } + } + + func headerValueAccount(_ header: Header) -> String { account("header.\(header.id.uuidString)") } + + func deleteSecrets() { + bearerToken = nil + hmacSecret = nil + mqttPassword = nil + for header in headers { + Keychain.set(nil, for: headerValueAccount(header)) + } + } + + // MARK: - URLs + + /// Where batches are POSTed. + var ingestURL: URL? { + switch kind { + case .pulso: + AppSettings.baseURL(from: url).flatMap { URL(string: $0.absoluteString + "/ingest") } + case .rest, .homeAssistant: + Self.httpURL(url) + case .mqtt, .file: + nil + } + } + + /// GET target for the catch-up hint, when enabled. + var latestURL: URL? { + guard catchUp else { return nil } + switch kind { + case .pulso: + return AppSettings.baseURL(from: url).flatMap { URL(string: $0.absoluteString + "/latest") } + case .rest: + return Self.httpURL(url).flatMap { url in + var base = url.absoluteString + while base.hasSuffix("/") { base.removeLast() } + return URL(string: base + "/latest") + } + default: + return nil + } + } + + /// A full endpoint URL, used exactly as typed (a trailing slash can be + /// significant to the server). + static func httpURL(_ raw: String) -> URL? { + var s = raw.trimmingCharacters(in: .whitespacesAndNewlines) + guard !s.isEmpty else { return nil } + if !s.contains("://") { s = "https://" + s } + guard let url = URL(string: s), url.host() != nil else { return nil } + return url + } + + /// Gets a copy of every batch. + var receivesBatches: Bool { + enabled && (isConfigured || wasConfigured) + } + + var isConfigured: Bool { + switch kind { + case .rest, .pulso, .homeAssistant: ingestURL != nil + case .mqtt: !mqttHost.trimmingCharacters(in: .whitespaces).isEmpty + case .file: true + } + } + + var summary: String { + switch kind { + case .rest, .pulso, .homeAssistant: ingestURL?.absoluteString ?? "no URL" + case .mqtt: "\(mqttHost):\(mqttPort) → \(mqttTopicPrefix)/" + case .file: folderName ?? "On My iPhone › Pulso › Pulso Export" + } + } +} diff --git a/Pulso/Destinations/MQTTPublisher.swift b/Pulso/Destinations/MQTTPublisher.swift new file mode 100644 index 0000000..6e51fdd --- /dev/null +++ b/Pulso/Destinations/MQTTPublisher.swift @@ -0,0 +1,355 @@ +import Foundation +import Network + +struct MQTTConfig: Sendable, Equatable { + var host: String + var port: UInt16 + var useTLS: Bool + var username: String? + var password: String? + var clientID: String + var keepAlive: UInt16 = 60 +} + +enum MQTTError: LocalizedError, Equatable { + case connectionRefused(returnCode: UInt8) + case timeout + case closed + case protocolViolation(String) + case network(String) + + var errorDescription: String? { + switch self { + case .connectionRefused(let returnCode): + return "MQTT connection refused (code \(returnCode))." + case .timeout: + return "MQTT connection timed out." + case .closed: + return "MQTT connection closed." + case .protocolViolation(let reason): + return "MQTT protocol violation: \(reason)" + case .network(let reason): + return "MQTT network error: \(reason)" + } + } +} + +enum MQTTPublisher { + /// Connects, publishes every message with QoS 1 and waits for each PUBACK, then sends DISCONNECT and closes. + /// A failed batch may have delivered a prefix; callers must tolerate duplicate deliveries on retry. + static func publish( + _ messages: [(topic: String, payload: Data, retain: Bool)], + config: MQTTConfig, + timeout: TimeInterval = 20 + ) async throws { + guard timeout.isFinite, timeout > 0 else { throw MQTTError.timeout } + guard !config.host.isEmpty, config.port != 0 else { + throw MQTTError.protocolViolation("Host and port must be nonempty and nonzero.") + } + try MQTTPacket.validateCredentials(config) + try MQTTPacket.validateString(config.clientID) + var connectLength = 10 + 2 + config.clientID.utf8.count + for credential in [config.username, config.password] { + if let credential, !credential.isEmpty { + try MQTTPacket.validateString(credential) + connectLength += 2 + credential.utf8.count + } + } + try MQTTPacket.validateLength(connectLength) + for message in messages { + try MQTTPacket.validateString(message.topic) + guard !message.topic.isEmpty, !message.topic.contains("#"), !message.topic.contains("+") else { + throw MQTTError.protocolViolation("PUBLISH requires a nonempty topic without wildcards.") + } + try MQTTPacket.validateLength(message.payload.count) + try MQTTPacket.validateLength(2 + message.topic.utf8.count + 2 + message.payload.count) + } + try Task.checkCancellation() + + let session = MQTTSession(config: config) + defer { session.cancel() } + try await withTaskCancellationHandler { + try await session.perform(.connect, timeout: timeout) + try await session.perform(.connack, sending: MQTTPacket.connect(config), timeout: timeout) + var packetID: UInt16 = 1 + for message in messages { + try Task.checkCancellation() + let packet = MQTTPacket.publish( + topic: message.topic, payload: message.payload, packetID: packetID, retain: message.retain + ) + try await session.perform(.puback(packetID), sending: packet, timeout: timeout) + packetID = packetID == .max ? 1 : packetID + 1 + } + try await session.perform(.disconnect, sending: MQTTPacket.disconnect(), timeout: timeout) + } onCancel: { + session.cancel() + } + } +} + +/// Encoding assumes validated input. The publisher checks the whole batch before +/// connecting so invalid lengths fail without publishing a prefix of that batch. +enum MQTTPacket { + private static let maximumLength = 268_435_455 + + static func encodeRemainingLength(_ length: Int) -> [UInt8] { + precondition((0 ... maximumLength).contains(length), "Invalid MQTT remaining length") + var remaining = length + var bytes: [UInt8] = [] + repeat { + var byte = UInt8(remaining % 128) + remaining /= 128 + if remaining > 0 { byte |= 0x80 } + bytes.append(byte) + } while remaining > 0 + return bytes + } + + static func connect(_ config: MQTTConfig, cleanSession: Bool = true) -> Data { + let username = config.username.flatMap { $0.isEmpty ? nil : $0 } + let password = config.password.flatMap { $0.isEmpty ? nil : $0 } + var flags: UInt8 = cleanSession ? 0x02 : 0 + if username != nil { flags |= 0x80 } + if password != nil { flags |= 0x40 } + var body = Data([0, 4, 0x4d, 0x51, 0x54, 0x54, 4, flags]) + appendUInt16(config.keepAlive, to: &body) + appendString(config.clientID, to: &body) + if let username { appendString(username, to: &body) } + if let password { appendString(password, to: &body) } + return frame(header: 0x10, body: body) + } + + static func publish(topic: String, payload: Data, packetID: UInt16, retain: Bool) -> Data { + precondition(packetID != 0, "MQTT packet IDs must be nonzero") + var body = Data() + appendString(topic, to: &body) + appendUInt16(packetID, to: &body) + body.append(payload) + return frame(header: retain ? 0x33 : 0x32, body: body) + } + + static func disconnect() -> Data { Data([0xe0, 0]) } + static func pingreq() -> Data { Data([0xc0, 0]) } + + static func parseFrame(_ buffer: Data) -> (type: UInt8, flags: UInt8, body: Data, consumed: Int)? { + try? checkedFrame(buffer) + } + + /// The receive path distinguishes malformed lengths from frames that need more bytes. + fileprivate static func checkedFrame(_ buffer: Data) throws -> (type: UInt8, flags: UInt8, body: Data, consumed: Int)? { + guard let header = buffer.first else { return nil } + var index = buffer.index(after: buffer.startIndex) + var length = 0 + var multiplier = 1 + for digit in 0 ..< 4 { + guard index < buffer.endIndex else { return nil } + let byte = buffer[index] + buffer.formIndex(after: &index) + length += Int(byte & 0x7f) * multiplier + if byte & 0x80 == 0 { + guard digit == 0 || byte != 0 else { + throw MQTTError.protocolViolation("Noncanonical remaining length.") + } + guard buffer.distance(from: index, to: buffer.endIndex) >= length else { return nil } + let end = buffer.index(index, offsetBy: length) + return (header >> 4, header & 0x0f, Data(buffer[index ..< end]), 2 + digit + length) + } + multiplier *= 128 + } + throw MQTTError.protocolViolation("Remaining length exceeds four bytes.") + } + + static func validateLength(_ length: Int) throws { + guard (0 ... maximumLength).contains(length) else { + throw MQTTError.protocolViolation("Packet exceeds the MQTT remaining-length limit.") + } + } + + static func validateCredentials(_ config: MQTTConfig) throws { + if let password = config.password, !password.isEmpty { + guard let username = config.username, !username.isEmpty else { + throw MQTTError.protocolViolation("A password requires a nonempty username.") + } + } + } + + fileprivate static func validateString(_ string: String) throws { + guard string.utf8.count <= Int(UInt16.max), !string.utf8.contains(0) else { + throw MQTTError.protocolViolation("String exceeds 65535 UTF-8 bytes or contains a null character.") + } + } + + private static func frame(header: UInt8, body: Data) -> Data { + var packet = Data([header]) + packet.append(contentsOf: encodeRemainingLength(body.count)) + packet.append(body) + return packet + } + + private static func appendUInt16(_ value: UInt16, to data: inout Data) { + data.append(contentsOf: [UInt8(value >> 8), UInt8(value & 0xff)]) + } + + private static func appendString(_ string: String, to data: inout Data) { + let bytes = Data(string.utf8) + precondition(bytes.count <= Int(UInt16.max), "MQTT string exceeds two-byte length") + appendUInt16(UInt16(bytes.count), to: &data) + data.append(bytes) + } +} + +/// All mutable state, including continuation ownership, lives on queue. +/// Serializing timeout, cancellation and Network callbacks makes each step complete once. +private final class MQTTSession: @unchecked Sendable { + enum Step: Equatable, Sendable { + case connect + case connack + case puback(UInt16) + case disconnect + } + + private let queue = DispatchQueue(label: "Pulso.MQTTPublisher") + private let connection: NWConnection + private var buffer = Data() + private var terminalError: MQTTError? + private var step: Step? + private var continuation: CheckedContinuation? + private var deadline: DispatchWorkItem? + private var generation: UInt64 = 0 + private var sent = false + private var acknowledged = false + + init(config: MQTTConfig) { + connection = NWConnection( + host: NWEndpoint.Host(config.host), port: NWEndpoint.Port(rawValue: config.port)!, + using: config.useTLS ? .tls : .tcp + ) + } + + func perform(_ next: Step, sending packet: Data? = nil, timeout: TimeInterval) async throws { + try Task.checkCancellation() + try await withCheckedThrowingContinuation { (pending: CheckedContinuation) in + queue.async { [self] in + if let terminalError { + pending.resume(throwing: terminalError) + return + } + precondition(continuation == nil) + continuation = pending + step = next + generation &+= 1 + let token = generation + sent = packet == nil + acknowledged = false + let timer = DispatchWorkItem { [weak self] in + guard let self, self.generation == token, self.continuation != nil else { return } + self.fail(.timeout) + } + deadline = timer + queue.asyncAfter(deadline: .now() + timeout, execute: timer) + if next == .connect { + start() + } else if let packet { + connection.send(content: packet, completion: .contentProcessed { [self] error in + if let error { fail(.network(error.localizedDescription)); return } + guard generation == token, continuation != nil else { return } + sent = true + if next == .disconnect || acknowledged { succeed() } + }) + } + } + } + } + + func cancel() { + queue.async { [self] in fail(.closed) } + } + + private func start() { + connection.stateUpdateHandler = { [self] state in + guard terminalError == nil else { return } + switch state { + case .ready: + guard step == .connect else { return } + succeed() + receive() + case .failed(let error), .waiting(let error): + fail(.network(error.localizedDescription)) + case .cancelled: + fail(.closed) + default: + break + } + } + connection.start(queue: queue) + } + + private func receive() { + connection.receive(minimumIncompleteLength: 1, maximumLength: 64 * 1024) { [self] data, _, complete, error in + guard terminalError == nil else { return } + if let data { buffer.append(data) } + do { + while let frame = try MQTTPacket.checkedFrame(buffer) { + buffer.removeFirst(frame.consumed) + try accept(type: frame.type, flags: frame.flags, body: frame.body) + } + } catch let error as MQTTError { + fail(error) + return + } catch { + fail(.protocolViolation(error.localizedDescription)) + return + } + if let error { fail(.network(error.localizedDescription)); return } + if complete { fail(.closed); return } + receive() + } + } + + private func accept(type: UInt8, flags: UInt8, body: Data) throws { + guard flags == 0 else { throw MQTTError.protocolViolation("Unexpected fixed-header flags.") } + if type == 13 { + guard body.isEmpty else { throw MQTTError.protocolViolation("Malformed PINGRESP.") } + return + } + guard !acknowledged else { throw MQTTError.protocolViolation("Duplicate acknowledgement.") } + switch step { + case .connack: + guard type == 2, body.count == 2, body[0] == 0 else { + throw MQTTError.protocolViolation("Expected a clean-session CONNACK.") + } + guard body[1] == 0 else { throw MQTTError.connectionRefused(returnCode: body[1]) } + case .puback(let packetID): + guard type == 4, body.count == 2, + (UInt16(body[0]) << 8 | UInt16(body[1])) == packetID else { + throw MQTTError.protocolViolation("Expected PUBACK for the current packet ID.") + } + default: + throw MQTTError.protocolViolation("Unexpected packet type \(type).") + } + acknowledged = true + if sent { succeed() } + } + + private func succeed() { + let pending = continuation + continuation = nil + step = nil + deadline?.cancel() + deadline = nil + pending?.resume() + } + + private func fail(_ error: MQTTError) { + guard terminalError == nil else { return } + terminalError = error + let pending = continuation + continuation = nil + step = nil + deadline?.cancel() + deadline = nil + connection.stateUpdateHandler = nil + connection.cancel() + pending?.resume(throwing: error) + } +} diff --git a/Pulso/Destinations/Transports.swift b/Pulso/Destinations/Transports.swift new file mode 100644 index 0000000..bc8af04 --- /dev/null +++ b/Pulso/Destinations/Transports.swift @@ -0,0 +1,333 @@ +import CryptoKit +import Foundation + +/// Delivers one outbox batch to one destination. A thrown error or a +/// non-2xx status means "not delivered": the batch stays queued. +protocol Transport: Sendable { + func deliver(_ batch: Batch) async throws -> UploadResult +} + +/// Who sent a batch; goes in the pulso.v2 envelope. +struct DeviceInfo: Codable, Equatable, Sendable { + var id: String + var name: String + var model: String + var systemVersion: String + var appVersion: String + var bundleId: String + + /// Filled in on the main actor at launch (UIDevice is main-actor only). + nonisolated(unsafe) static var current = DeviceInfo( + id: "unknown", name: "iPhone", model: "iPhone", systemVersion: "", + appVersion: "", bundleId: Bundle.main.bundleIdentifier ?? "" + ) +} + +/// Request bodies for each destination format. +enum BatchBody { + static let schema = "pulso.v2" + + private struct Envelope: Encodable { + let schema: String + let batchId: String + let type: String + let createdAt: Date + let sentAt: Date + let device: DeviceInfo + let items: [BatchElement] + } + + static func make(_ batch: Batch, format: Destination.BodyFormat, device: DeviceInfo, slim: Bool = false) throws -> Data { + let items = slim ? batch.elements.map(slimmed) : batch.elements + switch format { + case .array: + return try Wire.encodeElements(items) + case .ndjson: + return try ndjson(items) + case .envelope: + let envelope = Envelope( + schema: schema, batchId: batch.id.uuidString, type: batch.typeKey, + createdAt: batch.createdAt, sentAt: Date(), device: device, items: items + ) + return try Wire.fileEncoder().encode(envelope) + } + } + + static func ndjson(_ items: [BatchElement]) throws -> Data { + let encoder = Wire.bodyEncoder() + var out = Data() + for item in items { + out.append(try encoder.encode(item)) + out.append(0x0A) + } + return out + } + + /// The test body every HTTP destination accepts: an empty batch marked + /// as a test, so receivers can skip storing it. + static func test(device: DeviceInfo, format: Destination.BodyFormat) throws -> Data { + switch format { + case .array: + return Data("[]".utf8) + case .ndjson: + return Data() + case .envelope: + struct Test: Encodable { + let schema: String + let test = true + let sentAt: Date + let device: DeviceInfo + let items: [BatchElement] = [] + } + return try Wire.fileEncoder().encode(Test(schema: schema, sentAt: Date(), device: device)) + } + } + + /// Long arrays (route locations, ECG voltages, beats, linked sample + /// ids) replaced by their length — for destinations like Home Assistant + /// that want the event, not the raw series. + static func slimmed(_ element: BatchElement) -> BatchElement { + guard case .sample(var dto) = element, var details = dto.details else { return element } + for (key, value) in details { + if case .array(let values) = value, values.count > 50 { + details[key] = nil + details[key + "Count"] = .number(Double(values.count)) + } + } + dto.details = details + return .sample(dto) + } +} + +// MARK: - HTTP + +struct HTTPTransport: Transport { + let destination: Destination + let session: URLSession + + static func makeSession() -> URLSession { + let configuration = URLSessionConfiguration.ephemeral + configuration.timeoutIntervalForRequest = 30 + configuration.timeoutIntervalForResource = 120 + configuration.waitsForConnectivity = false // fail fast; the outbox owns retries + return URLSession(configuration: configuration) + } + + func deliver(_ batch: Batch) async throws -> UploadResult { + let body = try BatchBody.make( + batch, format: destination.format, device: DeviceInfo.current, + slim: destination.kind == .homeAssistant + ) + return try await post(body, extraHeaders: ["X-Pulso-Batch": batch.id.uuidString, "X-Pulso-Type": batch.typeKey]) + } + + func post(_ raw: Data, extraHeaders: [String: String] = [:]) async throws -> UploadResult { + guard let url = destination.ingestURL else { throw URLError(.badURL) } + let body = destination.gzip ? Gzip.compress(raw) : raw + var request = URLRequest(url: url) + request.httpMethod = "POST" + request.setValue(destination.format == .ndjson ? "application/x-ndjson" : "application/json", + forHTTPHeaderField: "Content-Type") + if destination.gzip { + request.setValue("gzip", forHTTPHeaderField: "Content-Encoding") + } + for (name, value) in extraHeaders { + request.setValue(value, forHTTPHeaderField: name) + } + for header in destination.headers where !header.name.trimmingCharacters(in: .whitespaces).isEmpty { + request.setValue(header.value, forHTTPHeaderField: header.name.trimmingCharacters(in: .whitespaces)) + } + if let token = destination.bearerToken, !token.isEmpty { + request.setValue("Bearer \(token)", forHTTPHeaderField: "Authorization") + } + if let secret = destination.hmacSecret, !secret.isEmpty { + // Signed over the bytes on the wire (after gzip), which is what + // the receiver can check before decoding anything. + request.setValue("sha256=" + Self.hmacHex(body, secret: secret), forHTTPHeaderField: "X-Signature") + } + let started = Date() + let (data, response) = try await session.upload(for: request, from: body, delegate: NoRedirects.shared) + return UploadResult( + status: (response as? HTTPURLResponse)?.statusCode ?? 0, + bodySnippet: String(decoding: data.prefix(300), as: UTF8.self), + duration: Date().timeIntervalSince(started) + ) + } + + static func hmacHex(_ body: Data, secret: String) -> String { + let mac = HMAC.authenticationCode(for: body, using: SymmetricKey(data: Data(secret.utf8))) + return mac.map { String(format: "%02x", $0) }.joined() + } +} + +/// Refuses redirects: URLSession would re-issue a redirected POST as a +/// body-less GET, and a 2xx to that would count as delivered. The 3xx +/// itself comes back instead, which the outbox treats as a failure. +final class NoRedirects: NSObject, URLSessionTaskDelegate { + static let shared = NoRedirects() + + func urlSession( + _ session: URLSession, task: URLSessionTask, willPerformHTTPRedirection response: HTTPURLResponse, + newRequest request: URLRequest + ) async -> URLRequest? { + nil + } +} + +// MARK: - MQTT + +/// One message per type per batch on / (a JSON array of +/// items), deletions on /deleted. Publishing every sample as its +/// own message would cost a broker round trip each. +struct MQTTTransport: Transport { + let destination: Destination + + var config: MQTTConfig { + MQTTConfig( + host: destination.mqttHost.trimmingCharacters(in: .whitespaces), + port: UInt16(clamping: destination.mqttPort), + useTLS: destination.mqttTLS, + username: destination.mqttUsername.isEmpty ? nil : destination.mqttUsername, + password: destination.mqttPassword, + // One per destination: a broker drops the older of two + // connections that share a client id. + clientID: "pulso-\(DeviceInfo.current.id.prefix(8))-\(destination.id.uuidString.prefix(8))" + ) + } + + func topic(_ suffix: String) -> String { + var prefix = destination.mqttTopicPrefix.trimmingCharacters(in: .whitespaces) + while prefix.hasSuffix("/") { prefix.removeLast() } + return prefix.isEmpty ? suffix : "\(prefix)/\(suffix)" + } + + func deliver(_ batch: Batch) async throws -> UploadResult { + var byType: [String: [BatchElement]] = [:] + var deleted: [String] = [] + for element in batch.elements { + switch element { + case .sample(let dto): byType[dto.type, default: []].append(element) + case .tombstone(let uuids): deleted += uuids + } + } + var messages: [(topic: String, payload: Data, retain: Bool)] = [] + for (type, items) in byType.sorted(by: { $0.key < $1.key }) { + messages.append((topic(type), try Wire.encodeElements(items), destination.mqttRetain)) + } + if !deleted.isEmpty { + messages.append((topic("deleted"), try Wire.encodeElements([.tombstone(deleted)]), false)) + } + let started = Date() + try await MQTTPublisher.publish(messages, config: config) + return UploadResult( + status: 200, bodySnippet: "published \(messages.count) message(s)", + duration: Date().timeIntervalSince(started) + ) + } + + func test() async throws { + let payload = try BatchBody.test(device: DeviceInfo.current, format: .envelope) + try await MQTTPublisher.publish([(topic("test"), payload, false)], config: config) + } +} + +// MARK: - Files + +/// Appends NDJSON lines to /pulso-.ndjson, one file per +/// day of export. The folder is the app's own (visible in Files under On My +/// iPhone › Pulso) or one the user picked, e.g. in iCloud Drive. +struct FileTransport: Transport { + let destination: Destination + + static var defaultFolder: URL { + let documents = FileManager.default.urls(for: .documentDirectory, in: .userDomainMask)[0] + return documents.appendingPathComponent("Pulso Export", isDirectory: true) + } + + func deliver(_ batch: Batch) async throws -> UploadResult { + let started = Date() + let lines = try BatchBody.ndjson(batch.elements) + let name = "pulso-\(Self.dayStamp(Date())).ndjson" + try withFolder { folder in + try FileManager.default.createDirectory(at: folder, withIntermediateDirectories: true) + try Self.append(lines, to: folder.appendingPathComponent(name)) + } + return UploadResult(status: 200, bodySnippet: "appended \(batch.elements.count) line(s) to \(name)", + duration: Date().timeIntervalSince(started)) + } + + func test() throws { + try withFolder { folder in + try FileManager.default.createDirectory(at: folder, withIntermediateDirectories: true) + try Data("Pulso can write here.\n".utf8).write(to: folder.appendingPathComponent("pulso-test.txt"), options: .atomic) + } + } + + private func withFolder(_ body: (URL) throws -> Void) throws { + guard let bookmark = destination.folderBookmark else { + return try body(Self.defaultFolder) + } + var stale = false + let folder: URL + do { + // A stale bookmark that still resolves keeps working; re-picking + // the folder refreshes it. + folder = try URL(resolvingBookmarkData: bookmark, bookmarkDataIsStale: &stale) + } catch { + throw FolderAccessError() + } + // Folders inside the app's own container need no scope, so a false + // here isn't fatal; the write itself reports a real permission error. + let scoped = folder.startAccessingSecurityScopedResource() + defer { if scoped { folder.stopAccessingSecurityScopedResource() } } + // Coordinated, so iCloud Drive sees the write as one change. + var coordinationError: NSError? + var bodyError: Error? + NSFileCoordinator().coordinate(writingItemAt: folder, options: .forMerging, error: &coordinationError) { url in + do { try body(url) } catch { bodyError = error } + } + if let error = coordinationError ?? bodyError { throw error } + } + + /// File destinations drain concurrently and may share a folder; one + /// lock keeps two appends from landing at the same offset. + private static let appendLock = NSLock() + + struct FolderAccessError: LocalizedError { + var errorDescription: String? { "Lost access to the export folder: choose it again in the destination's settings" } + } + + private static func append(_ data: Data, to file: URL) throws { + appendLock.lock() + defer { appendLock.unlock() } + if !FileManager.default.fileExists(atPath: file.path) { + try data.write(to: file, options: .atomic) + return + } + let handle = try FileHandle(forWritingTo: file) + defer { try? handle.close() } + try handle.seekToEnd() + try handle.write(contentsOf: data) + } + + private static func dayStamp(_ date: Date) -> String { + let formatter = DateFormatter() + formatter.calendar = Calendar(identifier: .gregorian) + formatter.locale = Locale(identifier: "en_US_POSIX") + formatter.dateFormat = "yyyy-MM-dd" + return formatter.string(from: date) + } +} + +extension Destination { + var transport: Transport? { + guard isConfigured else { return nil } + switch kind { + case .rest, .pulso, .homeAssistant: return HTTPTransport(destination: self, session: Self.session) + case .mqtt: return MQTTTransport(destination: self) + case .file: return FileTransport(destination: self) + } + } + + private static let session = HTTPTransport.makeSession() +} diff --git a/Pulso/Info.plist b/Pulso/Info.plist index 5d8a982..a276dd7 100644 --- a/Pulso/Info.plist +++ b/Pulso/Info.plist @@ -7,11 +7,17 @@ BGTaskSchedulerPermittedIdentifiers pulso.refresh + pulso.processing UIBackgroundModes fetch + processing + UIFileSharingEnabled + + LSSupportsOpeningDocumentsInPlace + NSAppTransportSecurity NSAllowsArbitraryLoads diff --git a/Pulso/Model/AppSettings.swift b/Pulso/Model/AppSettings.swift index b7255d3..4ea51da 100644 --- a/Pulso/Model/AppSettings.swift +++ b/Pulso/Model/AppSettings.swift @@ -2,6 +2,7 @@ import Foundation import Observation enum SettingsKeys { + /// The single server URL from before destinations; read once to migrate. static let endpoint = "server.endpoint" static let enabledTypes = "types.enabled" static let tokenAccount = "server-token" @@ -11,6 +12,7 @@ enum SettingsKeys { /// catch-up shortcut is disabled so the re-import cannot be defeated by /// an interleaved or resumed pass. static let reimportPending = "sync.reimportPending" + static let destinations = "destinations.v1" } /// User configuration. The @Observable instance drives the UI; the static @@ -19,30 +21,113 @@ enum SettingsKeys { @Observable @MainActor final class AppSettings { - var endpoint: String { - didSet { UserDefaults.standard.set(endpoint, forKey: SettingsKeys.endpoint) } - } - - var token: String { - didSet { Keychain.set(token.isEmpty ? nil : token, for: SettingsKeys.tokenAccount) } - } - private var enabled: [String: Bool] { didSet { UserDefaults.standard.set(enabled, forKey: SettingsKeys.enabledTypes) } } + /// Where data goes. Saved on every change; secrets are kept in the + /// Keychain by Destination itself. + var destinations: [Destination] { + didSet { Self.storeDestinations(destinations) } + } + init() { - endpoint = UserDefaults.standard.string(forKey: SettingsKeys.endpoint) ?? "" - token = Keychain.string(for: SettingsKeys.tokenAccount) ?? "" + Self.migrateLegacyEndpoint() enabled = Self.storedEnabled() + destinations = Self.currentDestinations() + // Rewrites the stored list once per launch, so a header value saved + // in plain text by an earlier build moves to the Keychain. + Self.storeDestinations(destinations) + } + + /// Before destinations, Pulso had one server URL and token. A non-empty + /// one becomes a "Pulso server" destination, once. + nonisolated static func migrateLegacyEndpoint() { + let defaults = UserDefaults.standard + guard defaults.data(forKey: SettingsKeys.destinations) == nil else { return } + var migrated: [Destination] = [] + if let raw = defaults.string(forKey: SettingsKeys.endpoint), baseURL(from: raw) != nil { + var destination = Destination(name: "Pulso server", kind: .pulso) + destination.url = raw + destination.bearerToken = Keychain.string(for: SettingsKeys.tokenAccount) + migrated.append(destination) + } + storeDestinations(migrated) + } + + /// Header values go to the Keychain (written only when they change); + /// the list is saved with them blank. Values of header rows that were + /// removed are deleted from the Keychain. + nonisolated static func storeDestinations(_ destinations: [Destination]) { + let previous = storedDestinations() + var stripped = destinations + for index in stripped.indices { + // Once true, stays true: the copy being edited may never have + // seen the flag the saved copy got. + let saved = previous.first { $0.id == stripped[index].id } + stripped[index].wasConfigured = stripped[index].isConfigured || stripped[index].wasConfigured + || saved?.wasConfigured == true + let destination = stripped[index] + let kept = Set(destination.headers.map(\.id)) + for old in saved?.headers ?? [] where !kept.contains(old.id) { + Keychain.set(nil, for: destination.headerValueAccount(old)) + } + for headerIndex in stripped[index].headers.indices { + let header = stripped[index].headers[headerIndex] + let account = destination.headerValueAccount(header) + if Keychain.string(for: account) ?? "" != header.value { + Keychain.set(header.value, for: account) + } + stripped[index].headers[headerIndex].value = "" + } + } + if let data = try? JSONEncoder().encode(stripped) { + UserDefaults.standard.set(data, forKey: SettingsKeys.destinations) + } + } + + nonisolated static func currentDestinations() -> [Destination] { + var destinations = storedDestinations() + for index in destinations.indices { + for headerIndex in destinations[index].headers.indices { + let header = destinations[index].headers[headerIndex] + // A value saved in the list by an earlier build is kept until + // the next save moves it to the Keychain. + destinations[index].headers[headerIndex].value = + Keychain.string(for: destinations[index].headerValueAccount(header)) ?? header.value + } + } + return destinations + } + + /// The stored list as saved; an entry that can't be decoded (written by a + /// newer version) is skipped rather than emptying the whole list. + nonisolated private static func storedDestinations() -> [Destination] { + struct Lossy: Decodable { + let value: Destination? + init(from decoder: Decoder) throws { value = try? Destination(from: decoder) } + } + guard let data = UserDefaults.standard.data(forKey: SettingsKeys.destinations), + let entries = try? JSONDecoder().decode([Lossy].self, from: data) else { return [] } + return entries.compactMap(\.value) + } + + /// Destinations that get a copy of every batch. + nonisolated static func receivingDestinations() -> [Destination] { + currentDestinations().filter(\.receivesBatches) + } + + /// Identity of a set of receiving destinations for catch-up: hints from + /// /latest are only valid for the exact set they were gathered from. + nonisolated static func catchUpKey(_ destinations: [Destination]) -> String { + destinations.map { "\($0.id.uuidString)|\($0.latestURL?.absoluteString ?? "-")" } + .sorted() + .joined(separator: ",") } func isEnabled(_ key: String) -> Bool { enabled[key] ?? true } func setEnabled(_ key: String, _ value: Bool) { enabled[key] = value } - var baseURL: URL? { Self.baseURL(from: endpoint) } - var healthURL: URL? { baseURL.flatMap { URL(string: $0.absoluteString + "/health") } } - // MARK: - Shared parsing / thread-safe accessors nonisolated static func baseURL(from raw: String) -> URL? { @@ -54,19 +139,6 @@ final class AppSettings { return url } - /// Where to POST, resolved at send time. nil = not configured yet. - nonisolated static func currentTarget() -> (url: URL, token: String?)? { - guard let base = currentBase(), - let url = URL(string: base.url.absoluteString + "/ingest") else { return nil } - return (url, base.token) - } - - nonisolated static func currentBase() -> (url: URL, token: String?)? { - let raw = UserDefaults.standard.string(forKey: SettingsKeys.endpoint) ?? "" - guard let base = baseURL(from: raw) else { return nil } - return (base, Keychain.string(for: SettingsKeys.tokenAccount)) - } - nonisolated static func currentEnabled(_ key: String) -> Bool { storedEnabled()[key] ?? true } diff --git a/Pulso/Model/AppStatus.swift b/Pulso/Model/AppStatus.swift index e47ed8b..9850da5 100644 --- a/Pulso/Model/AppStatus.swift +++ b/Pulso/Model/AppStatus.swift @@ -1,6 +1,15 @@ import Foundation import Observation +struct DestinationStatus: Codable, Equatable, Sendable { + var pending = 0 + var nextRetry: Date? + var lastResponse: String? + var lastSuccess: Date? + var lastFailure: Date? + var delivered = 0 +} + struct TypeStatus: Codable, Equatable, Sendable { var lastChecked: Date? var lastDelivered: Date? @@ -21,9 +30,15 @@ final class AppStatus { var outboxNextRetry: Date? var lastResponse: String? var recentLog: [LogEntry] = [] + var destinationStatus: [UUID: DestinationStatus] = [:] + var lastSyncStarted: Date? + var lastSyncFinished: Date? + var lastSyncSummary: String? private static let statusKey = "status.types" private static let responseKey = "status.lastResponse" + private static let destinationsKey = "status.destinations" + private static let lastSyncKey = "status.lastSync" private let maxLogEntries = 200 func load() { @@ -32,6 +47,82 @@ final class AppStatus { typeStatus = decoded } lastResponse = UserDefaults.standard.string(forKey: Self.responseKey) + if let data = UserDefaults.standard.data(forKey: Self.destinationsKey), + let decoded = try? JSONDecoder().decode([UUID: DestinationStatus].self, from: data) { + destinationStatus = decoded + } + if let data = UserDefaults.standard.data(forKey: Self.lastSyncKey), + let decoded = try? JSONDecoder().decode(LastSync.self, from: data) { + lastSyncStarted = decoded.started + lastSyncFinished = decoded.finished + lastSyncSummary = decoded.summary + } + } + + /// Persisted so App Intents (which may run with the UI never shown) + /// and the next launch see the same answer. + struct LastSync: Codable, Sendable { + var started: Date? + var finished: Date? + var summary: String? + } + + nonisolated static func storedLastSync() -> LastSync? { + UserDefaults.standard.data(forKey: lastSyncKey).flatMap { try? JSONDecoder().decode(LastSync.self, from: $0) } + } + + nonisolated static func storedDestinationStatus() -> [UUID: DestinationStatus] { + UserDefaults.standard.data(forKey: destinationsKey) + .flatMap { try? JSONDecoder().decode([UUID: DestinationStatus].self, from: $0) } ?? [:] + } + + func recordSyncStarted() { + lastSyncStarted = Date() + saveLastSync() + } + + func recordSyncFinished(_ summary: String) { + lastSyncFinished = Date() + lastSyncSummary = summary + saveLastSync() + } + + private func saveLastSync() { + let value = LastSync(started: lastSyncStarted, finished: lastSyncFinished, summary: lastSyncSummary) + if let data = try? JSONEncoder().encode(value) { + UserDefaults.standard.set(data, forKey: Self.lastSyncKey) + } + } + + func recordDestinationDelivery(_ id: UUID, count: Int, response: String) { + updateDestination(id) { + $0.lastSuccess = Date() + $0.lastResponse = response + $0.delivered += count + } + } + + func recordDestinationFailure(_ id: UUID, response: String) { + updateDestination(id) { + $0.lastFailure = Date() + $0.lastResponse = response + } + } + + func setDestinationOutbox(_ id: UUID, pending: Int, nextRetry: Date?) { + updateDestination(id) { + $0.pending = pending + $0.nextRetry = nextRetry + } + } + + private func updateDestination(_ id: UUID, _ mutate: (inout DestinationStatus) -> Void) { + var value = destinationStatus[id] ?? DestinationStatus() + mutate(&value) + destinationStatus[id] = value + if let data = try? JSONEncoder().encode(destinationStatus) { + UserDefaults.standard.set(data, forKey: Self.destinationsKey) + } } private func save() { @@ -70,6 +161,11 @@ final class AppStatus { func resetCounters() { typeStatus = [:] + destinationStatus = destinationStatus.mapValues { status in + var reset = DestinationStatus() + reset.pending = status.pending + return reset + } lastResponse = nil save() } diff --git a/Pulso/Support/Keychain.swift b/Pulso/Support/Keychain.swift index cd92e0f..5370aca 100644 --- a/Pulso/Support/Keychain.swift +++ b/Pulso/Support/Keychain.swift @@ -15,14 +15,23 @@ enum Keychain { return String(data: data, encoding: .utf8) } - /// nil or empty deletes the item. + /// nil or empty deletes the item. An existing item is updated in place, + /// so a concurrent reader never sees it missing mid-write. static func set(_ value: String?, for account: String) { - SecItemDelete(base(account) as CFDictionary) - guard let value, !value.isEmpty else { return } - var attributes = base(account) - attributes[kSecValueData as String] = Data(value.utf8) - attributes[kSecAttrAccessible as String] = kSecAttrAccessibleAfterFirstUnlockThisDeviceOnly - SecItemAdd(attributes as CFDictionary, nil) + guard let value, !value.isEmpty else { + SecItemDelete(base(account) as CFDictionary) + return + } + let data = Data(value.utf8) + let update: [String: Any] = [ + kSecValueData as String: data, + kSecAttrAccessible as String: kSecAttrAccessibleAfterFirstUnlockThisDeviceOnly, + ] + if SecItemUpdate(base(account) as CFDictionary, update as CFDictionary) == errSecItemNotFound { + var attributes = base(account) + attributes.merge(update) { _, new in new } + SecItemAdd(attributes as CFDictionary, nil) + } } private static func base(_ account: String) -> [String: Any] { diff --git a/Pulso/Sync/Outbox.swift b/Pulso/Sync/Outbox.swift index 2e76b9a..43437ba 100644 --- a/Pulso/Sync/Outbox.swift +++ b/Pulso/Sync/Outbox.swift @@ -1,22 +1,32 @@ import Foundation -/// Durable delivery queue. Every batch is written to disk *before* the first -/// upload attempt and deleted only on a server 2xx — at which point the -/// batch's anchor (if any) is persisted. Files drain strictly in enqueue -/// order and draining stops at the first failure, so anchors can never -/// advance past an undelivered batch. At-least-once delivery; the server -/// dedupes by sample uuid. +/// Durable delivery queue for one destination. Every batch is written to +/// disk *before* the first delivery attempt and deleted only once the +/// destination accepts it (a 2xx, a broker PUBACK, a completed file write). +/// Files drain strictly in enqueue order and draining stops at the first +/// failure. At-least-once delivery; receivers dedupe by sample uuid. +/// +/// With `anchors`, a batch's anchor is persisted when it is accepted (the +/// original single-server design). OutboxHub passes nil: with several +/// destinations SyncEngine persists anchors once a batch is on disk in +/// every queue, so no destination can hold back another. actor Outbox { static let baseRetryInterval: TimeInterval = 60 static let maxRetryInterval: TimeInterval = 3600 private let directory: URL - private let anchors: AnchorStore - private let uploader: Uploading - private let target: @Sendable () -> (url: URL, token: String?)? + private let anchors: AnchorStore? + private let label: String + private let transport: @Sendable () -> Transport? private let log: LogStore private let status: AppStatus private let maxBatchBytes: Int + private let maxItems: Int? + private let destinationID: UUID? + /// Per-type delivery counts come from one destination only, or two + /// destinations would double them. + private let countsTowardTypes: @Sendable () -> Bool + private let sequenceKey: String private var sequence: Int private var failureCount = 0 private var nextAttempt = Date.distantPast @@ -24,29 +34,58 @@ actor Outbox { init( directory: URL, - anchors: AnchorStore, - uploader: Uploading, - target: @escaping @Sendable () -> (url: URL, token: String?)?, + anchors: AnchorStore?, + label: String = "outbox", + transport: @escaping @Sendable () -> Transport?, log: LogStore, status: AppStatus, - maxBatchBytes: Int = Wire.maxBatchBytes + maxBatchBytes: Int = Wire.maxBatchBytes, + maxItems: Int? = nil, + destinationID: UUID? = nil, + countsTowardTypes: @escaping @Sendable () -> Bool = { true } ) { self.directory = directory self.anchors = anchors - self.uploader = uploader - self.target = target + self.label = label + self.transport = transport self.log = log self.status = status self.maxBatchBytes = maxBatchBytes - sequence = UserDefaults.standard.integer(forKey: SettingsKeys.outboxSequence) + self.maxItems = maxItems + self.destinationID = destinationID + self.countsTowardTypes = countsTowardTypes + sequenceKey = destinationID.map { "\(SettingsKeys.outboxSequence).\($0.uuidString)" } ?? SettingsKeys.outboxSequence + sequence = UserDefaults.standard.integer(forKey: sequenceKey) try? FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) } + /// The original single-server shape: gzip the bare array and POST it. + init( + directory: URL, + anchors: AnchorStore, + uploader: Uploading, + target: @escaping @Sendable () -> (url: URL, token: String?)?, + log: LogStore, + status: AppStatus, + maxBatchBytes: Int = Wire.maxBatchBytes + ) { + self.init( + directory: directory, anchors: anchors, + transport: { target().map { LegacyUploadTransport(uploader: uploader, url: $0.url, token: $0.token) } }, + log: log, status: status, maxBatchBytes: maxBatchBytes + ) + } + // MARK: - Enqueue func enqueue(typeKey: String, elements: [BatchElement], anchor: Data?) throws { guard !elements.isEmpty else { return } - let slices = try BatchSlicer.slice(elements, maxBytes: maxBatchBytes) + var slices = try BatchSlicer.slice(elements, maxBytes: maxBatchBytes) + if let maxItems { + slices = slices.flatMap { slice in + stride(from: 0, to: slice.count, by: maxItems).map { Array(slice[$0 ..< min($0 + maxItems, slice.count)]) } + } + } for (index, slice) in slices.enumerated() { let batch = Batch( id: UUID(), @@ -57,7 +96,7 @@ actor Outbox { ) let data = try Wire.fileEncoder().encode(batch) sequence += 1 - UserDefaults.standard.set(sequence, forKey: SettingsKeys.outboxSequence) + UserDefaults.standard.set(sequence, forKey: sequenceKey) let name = String(format: "batch-%012d-%@.json", sequence, String(batch.id.uuidString.prefix(8))) try data.write(to: directory.appendingPathComponent(name), options: .atomic) } @@ -83,13 +122,16 @@ actor Outbox { defer { Task { await self.refreshPendingStatus() } } var files = pendingFiles() guard !files.isEmpty else { return } - guard let target = target() else { - await log.info("outbox: \(files.count) batch(es) waiting — no server configured") + guard transport() != nil else { + await log.info("\(label): \(files.count) batch(es) waiting — destination not configured") return } if !force && Date() < nextAttempt { return } while let file = files.first, !Task.isCancelled { + // Resolved per batch so disabling or editing the destination + // takes effect mid-drain. + guard let transport = transport() else { return } files.removeFirst() let batch: Batch do { @@ -97,26 +139,30 @@ actor Outbox { } catch { // A batch file we can't read would poison the queue forever; set it aside. quarantine(file) - await log.error("outbox: unreadable batch \(file.lastPathComponent) set aside (\(error.localizedDescription))") + await log.error("\(label): unreadable batch \(file.lastPathComponent) set aside (\(error.localizedDescription))") continue } do { - let body = try Wire.encodeElements(batch.elements) - let result = try await uploader.send(Gzip.compress(body), gzipped: true, to: target.url, token: target.token) + let result = try await transport.deliver(batch) guard result.isSuccess else { await recordFailure("HTTP \(result.status) \(result.bodySnippet.prefix(120))", status: result.status) return } try? FileManager.default.removeItem(at: file) if let anchorData = batch.anchor { - anchors.saveRaw(anchorData, for: batch.typeKey) + anchors?.saveRaw(anchorData, for: batch.typeKey) } failureCount = 0 nextAttempt = .distantPast - let response = "HTTP \(result.status) · \(result.bodySnippet.prefix(80))" - await status.recordDelivery(typeKey: batch.typeKey, count: batch.sampleCount, response: response) + let response = "\(result.status) · \(result.bodySnippet.prefix(80))" + if countsTowardTypes() { + await status.recordDelivery(typeKey: batch.typeKey, count: batch.sampleCount, response: response) + } + if let destinationID { + await status.recordDestinationDelivery(destinationID, count: batch.sampleCount, response: response) + } await log.info( - "delivered \(batch.sampleCount) \(batch.typeKey) sample(s) — HTTP \(result.status) in \(Int(result.duration * 1000))ms" + "\(label): delivered \(batch.sampleCount) \(batch.typeKey) sample(s) — \(result.status) in \(Int(result.duration * 1000))ms" ) } catch { await recordFailure(error.localizedDescription, status: nil) @@ -133,9 +179,15 @@ actor Outbox { ) nextAttempt = Date(timeIntervalSinceNow: delay) var hint = "" - if statusCode == 401 { hint = " — check the bearer token in Settings" } - await status.setLastResponse(String(reason.prefix(160))) - await log.warn("delivery failed: \(reason)\(hint) — retrying in \(Int(delay))s") + if statusCode == 401 { hint = " — check the token / signature secret in Settings" } + let text = String(reason.prefix(160)) + if let destinationID { + await status.recordDestinationFailure(destinationID, response: text) + } + if countsTowardTypes() { + await status.setLastResponse(text) + } + await log.warn("\(label): delivery failed: \(reason)\(hint) — retrying in \(Int(delay))s") } private func quarantine(_ file: URL) { @@ -160,7 +212,12 @@ actor Outbox { func refreshPendingStatus() async { let count = pendingFiles().count - await status.setOutbox(pending: count, nextRetry: failureCount > 0 && count > 0 ? nextAttempt : nil) + let retry = failureCount > 0 && count > 0 ? nextAttempt : nil + if let destinationID { + await status.setDestinationOutbox(destinationID, pending: count, nextRetry: retry) + } else { + await status.setOutbox(pending: count, nextRetry: retry) + } } func removeAll() async { @@ -172,3 +229,15 @@ actor Outbox { await refreshPendingStatus() } } + +/// The original POST: gzipped bare array through an `Uploading`. +struct LegacyUploadTransport: Transport { + let uploader: Uploading + let url: URL + let token: String? + + func deliver(_ batch: Batch) async throws -> UploadResult { + let body = try Wire.encodeElements(batch.elements) + return try await uploader.send(Gzip.compress(body), gzipped: true, to: url, token: token) + } +} diff --git a/Pulso/Sync/OutboxHub.swift b/Pulso/Sync/OutboxHub.swift new file mode 100644 index 0000000..f5def5d --- /dev/null +++ b/Pulso/Sync/OutboxHub.swift @@ -0,0 +1,141 @@ +import Foundation + +/// One Outbox per destination, each in its own folder. A batch is written +/// to every enabled destination's queue before enqueue returns; each queue +/// then drains on its own schedule, so a destination that is down only +/// delays itself. +actor OutboxHub { + enum EnqueueError: LocalizedError { + case noDestination + /// The receiving destinations changed since the catch-up hints were + /// fetched; the batch would skip history a new destination needs. + case destinationsChanged + + var errorDescription: String? { + switch self { + case .noDestination: "no enabled destination is configured" + case .destinationsChanged: "destinations changed during catch-up" + } + } + } + + private let root: URL + private let log: LogStore + private let status: AppStatus + private let maxBatchBytes: Int + private let destinations: @Sendable () -> [Destination] + private let makeTransport: @Sendable (Destination) -> Transport? + private var outboxes: [UUID: Outbox] = [:] + + init( + root: URL, log: LogStore, status: AppStatus, + maxBatchBytes: Int = Wire.maxBatchBytes, + destinations: @escaping @Sendable () -> [Destination] = { AppSettings.currentDestinations() }, + transport: @escaping @Sendable (Destination) -> Transport? = { $0.transport } + ) { + self.root = root + self.log = log + self.status = status + self.maxBatchBytes = maxBatchBytes + self.destinations = destinations + makeTransport = transport + try? FileManager.default.createDirectory(at: root, withIntermediateDirectories: true) + } + + private func outbox(for destination: Destination) -> Outbox { + if let existing = outboxes[destination.id] { return existing } + let id = destination.id + let destinations = destinations + let makeTransport = makeTransport + let outbox = Outbox( + directory: root.appendingPathComponent(id.uuidString, isDirectory: true), + anchors: nil, + label: destination.name, + // Resolved at send time so edits (URL, token, …) apply to + // batches already queued. + transport: { destinations().first { $0.id == id && $0.enabled }.flatMap(makeTransport) }, + log: log, + status: status, + maxBatchBytes: maxBatchBytes, + maxItems: destination.maxItemsPerBatch, + destinationID: id, + // Per-type counts follow the first deliverable destination, + // decided per batch so removing it hands the role on. + countsTowardTypes: { destinations().first { $0.enabled && $0.isConfigured }?.id == id } + ) + outboxes[id] = outbox + return outbox + } + + /// Every destination that receives batches, configured right now or + /// not: one whose URL is being edited keeps queueing and delivers once + /// it is valid again, instead of missing the batches synced meanwhile. + private var active: [(destination: Destination, outbox: Outbox)] { + destinations().filter(\.receivesBatches).map { ($0, outbox(for: $0)) } + } + + /// True when at least one destination can deliver right now. + func hasActiveDestination() -> Bool { + destinations().contains { $0.enabled && $0.isConfigured } + } + + /// Writes the elements to every enabled destination's queue. Throws if + /// none can deliver or a write fails, in which case the caller must not + /// advance its anchor. + /// With `catchUpKey`, refuses unless the destinations written to are + /// exactly the set the catch-up hints came from (checked against the + /// same list it writes to). + func enqueue(typeKey: String, elements: [BatchElement], catchUpKey: String? = nil) async throws { + // Every decision comes from this one snapshot: a second read could + // see a deliverable destination this list doesn't write to. + let targets = active + guard targets.contains(where: { $0.destination.isConfigured }) else { throw EnqueueError.noDestination } + if let catchUpKey, AppSettings.catchUpKey(targets.map(\.destination)) != catchUpKey { + throw EnqueueError.destinationsChanged + } + for target in targets { + try await target.outbox.enqueue(typeKey: typeKey, elements: elements, anchor: nil) + } + } + + func drain(force: Bool) async { + let targets = active.map(\.outbox) + await withTaskGroup(of: Void.self) { group in + for outbox in targets { + group.addTask { await outbox.drain(force: force) } + } + } + } + + func drain(destination id: UUID, force: Bool) async { + guard let target = active.first(where: { $0.destination.id == id }) else { return } + await target.outbox.drain(force: force) + } + + func pendingCount() async -> Int { + var total = 0 + for target in active { + total += await target.outbox.pendingCount() + } + return total + } + + func refreshPendingStatus() async { + for target in active { + await target.outbox.refreshPendingStatus() + } + await status.setOutbox(pending: await pendingCount(), nextRetry: nil) + } + + func removeAll() async { + for target in active { + await target.outbox.removeAll() + } + } + + /// Drops the queue of a destination that was deleted. + func forget(_ id: UUID) { + outboxes[id] = nil + try? FileManager.default.removeItem(at: root.appendingPathComponent(id.uuidString, isDirectory: true)) + } +} diff --git a/Pulso/Sync/SyncEngine.swift b/Pulso/Sync/SyncEngine.swift index 93bb693..61221f8 100644 --- a/Pulso/Sync/SyncEngine.swift +++ b/Pulso/Sync/SyncEngine.swift @@ -8,6 +8,8 @@ enum SyncReason: CustomStringConvertible, Sendable { case backgroundRefresh case settingsChanged case reimport + case intent(String) + case processing var description: String { switch self { @@ -17,6 +19,8 @@ enum SyncReason: CustomStringConvertible, Sendable { case .backgroundRefresh: "background refresh" case .settingsChanged: "settings changed" case .reimport: "full re-import" + case .intent(let name): "shortcut: \(name)" + case .processing: "background processing" } } } @@ -31,6 +35,38 @@ enum CatchUp { /// future-dated server timestamp would otherwise produce an empty /// bounded query that persists a store-wide anchor and permanently /// skips the type's history. + /// Combines several servers' /latest answers: a type counts only if + /// every server has it, at the earliest of their timestamps. + static func earliest(_ answers: [[String: String]]) -> [String: String] { + guard var merged = answers.first else { return [:] } + for answer in answers.dropFirst() { + for (type, value) in merged { + guard let other = answer[type] else { + merged[type] = nil + continue + } + // A timestamp that doesn't parse says nothing safe about + // what that server holds: no catch-up for the type. + guard let a = parse(value), let b = parse(other) else { + merged[type] = nil + continue + } + if b < a { + merged[type] = other + } + } + } + return merged + } + + private static func parse(_ raw: String) -> Date? { + let strict = ISO8601DateFormatter() + strict.formatOptions = [.withInternetDateTime] + let fractional = ISO8601DateFormatter() + fractional.formatOptions = [.withInternetDateTime, .withFractionalSeconds] + return strict.date(from: raw) ?? fractional.date(from: raw) + } + static func cutoffs(fromLatest latest: [String: String], now: Date = Date()) -> [String: Date] { let strict = ISO8601DateFormatter() strict.formatOptions = [.withInternetDateTime] @@ -43,23 +79,23 @@ enum CatchUp { } } -/// Per enabled type: run an anchored query from the last anchor, hand the -/// results to the outbox, and let the outbox persist the new anchor once the -/// server ACKs. An in-memory anchor cache leads the persisted one so -/// overlapping triggers don't re-read samples that are already queued; after -/// a crash the cache is gone, the persisted anchor wins, and any re-read -/// samples dedupe server-side by uuid. +/// Per enabled type: run an anchored query from the last anchor, write the +/// results into every destination's durable queue, then persist the new +/// anchor. The data is on disk before the anchor moves, and each queue +/// deletes a batch only once its destination accepts it; a crash between +/// the two re-reads samples that receivers dedupe by uuid. actor SyncEngine { static let queryLimit = 5000 private let healthStore: HKHealthStore private let queries: HealthQueries private let anchors: AnchorStore - private let outbox: Outbox + private let outbox: OutboxHub private let log: LogStore private let status: AppStatus /// GET /latest from the configured server; nil when unavailable. - private let latestHints: @Sendable () async -> [String: String]? + /// GET /latest from the given destinations; nil when unavailable. + private let latestHints: @Sendable ([Destination]) async -> [String: String]? private var anchorCache: [String: HKQueryAnchor] = [:] /// Snapshot state as last queued; survives in memory between passes. private var snapshotCache: [String: SnapshotState] = [:] @@ -71,13 +107,15 @@ actor SyncEngine { private var authBlocked: Set = [] /// GET /latest result (including failure) cached briefly so background /// wakes don't pay a network round-trip per pass. - private var latestFetch: (at: Date, hints: [String: String]?)? + /// Keyed by the destinations that answered: a hint is only valid for + /// the exact set of destinations it was gathered from. + private var latestFetch: (at: Date, destinations: String, hints: [String: String]?)? private static let latestCacheInterval: TimeInterval = 15 * 60 init( - healthStore: HKHealthStore, anchors: AnchorStore, outbox: Outbox, + healthStore: HKHealthStore, anchors: AnchorStore, outbox: OutboxHub, log: LogStore, status: AppStatus, - latestHints: @escaping @Sendable () async -> [String: String]? = { nil } + latestHints: @escaping @Sendable ([Destination]) async -> [String: String]? = { _ in nil } ) { self.healthStore = healthStore queries = HealthQueries(store: healthStore) @@ -103,11 +141,12 @@ actor SyncEngine { } private func performSync(_ reason: SyncReason, only keys: Set?) async { - guard AppSettings.currentTarget() != nil else { - await log.info("sync skipped (\(reason)): no server configured") + guard await outbox.hasActiveDestination() else { + await log.info("sync skipped (\(reason)): no destination configured") return } await status.setSyncing(true) + await status.recordSyncStarted() // Types this OS refused at authorization would only fail their query. let refused = HealthAuthorization.lastRejected() let enabled = AppSettings.currentEnabledTypes().filter { @@ -117,9 +156,9 @@ actor SyncEngine { // A user-visible pass may follow a permission grant — retry types // that were failing. Background wakes keep them excluded. switch reason { - case .manual, .foreground, .settingsChanged, .reimport: + case .manual, .foreground, .settingsChanged, .reimport, .intent: authBlocked.removeAll() - case .observer, .backgroundRefresh: + case .observer, .backgroundRefresh, .processing: break } @@ -142,9 +181,11 @@ actor SyncEngine { // an interleaved or resumed pass would otherwise catch-up against // the fully-populated server and defeat the re-import. var cutoffs: [String: Date] = [:] + var catchUpKey: String? let reimporting = UserDefaults.standard.bool(forKey: SettingsKeys.reimportPending) if !reimporting && !missing.isEmpty { - if let latest = await cachedLatestHints() { + if let (latest, key) = await cachedLatestHints() { + catchUpKey = key let parsed = CatchUp.cutoffs(fromLatest: latest) if parsed.count < latest.count { await log.warn("server /latest: \(latest.count - parsed.count) unparseable timestamp(s) ignored") @@ -160,7 +201,7 @@ actor SyncEngine { var anyFailed = false for type in enabled { do { - queued += try await sync(type, anchor: resolvedAnchors[type.key], cutoffs: cutoffs) + queued += try await sync(type, anchor: resolvedAnchors[type.key], cutoffs: cutoffs, catchUpKey: catchUpKey) } catch { anyFailed = true authBlocked.insert(type.key) @@ -168,15 +209,13 @@ actor SyncEngine { } } // Snapshots have no observer of their own; they ride along on every - // full pass. - if keys == nil { - for source in AppSettings.currentEnabledSnapshots() { - do { - queued += try await syncSnapshot(source) - } catch { - anyFailed = true - await log.error("\(source.key): \(error.localizedDescription)") - } + // full pass, or when a shortcut names them. + for source in AppSettings.currentEnabledSnapshots() where keys?.contains(source.key) ?? true { + do { + queued += try await syncSnapshot(source) + } catch { + anyFailed = true + await log.error("\(source.key): \(error.localizedDescription)") } } if reimporting && !anyFailed && keys == nil { @@ -185,26 +224,139 @@ actor SyncEngine { } let force: Bool switch reason { - case .manual, .foreground, .settingsChanged, .reimport: force = true + case .manual, .foreground, .settingsChanged, .reimport, .intent, .processing: force = true case .observer, .backgroundRefresh: force = false } await outbox.drain(force: force) + let pending = await outbox.pendingCount() + let failed = authBlocked.sorted() + var summary = "\(reason): queued \(queued) sample(s), \(pending) batch(es) waiting" + if !failed.isEmpty { + summary += ", \(failed.count) type(s) failing (\(failed.prefix(5).joined(separator: ", ")))" + } + await status.recordSyncFinished(summary) await status.setSyncing(false) if queued > 0 { await log.info("sync (\(reason)): queued \(queued) sample(s)") } } - private func cachedLatestHints() async -> [String: String]? { - if let cached = latestFetch, Date().timeIntervalSince(cached.at) < Self.latestCacheInterval { - return cached.hints + /// Re-sends what HealthKit holds between `start` and `end` (open ends + /// mean all of it) without touching anchors: the backfill shortcut. + /// Serialized with sync passes. + func resend(only keys: Set?, start: Date?, end: Date?, reason: SyncReason) async -> String { + let previous = chain + let task = Task { () -> String in + await previous?.value + return await self.performResend(only: keys, start: start, end: end, reason: reason) } - let hints = await latestHints() - latestFetch = (Date(), hints) - return hints + chain = Task { _ = await task.value } + return await task.value } - private func sync(_ type: SyncedType, anchor resolvedAnchor: HKQueryAnchor?, cutoffs: [String: Date]) async throws -> Int { + private func performResend(only keys: Set?, start: Date?, end: Date?, reason: SyncReason) async -> String { + guard await outbox.hasActiveDestination() else { return "No destination is configured." } + await status.setSyncing(true) + await status.recordSyncStarted() + let refused = HealthAuthorization.lastRejected() + let types = AppSettings.currentEnabledTypes().filter { + !refused.contains($0.sampleType.identifier) && (keys?.contains($0.key) ?? true) + } + let predicate = (start != nil || end != nil) ? HKQuery.predicateForSamples(withStart: start, end: end) : nil + var queued = 0 + var failed: [String] = [] + for type in types { + do { + var anchor: HKQueryAnchor? + while true { + let descriptor = HKAnchoredObjectQueryDescriptor( + predicates: [.sample(type: type.sampleType, predicate: predicate)], + anchor: anchor, limit: type.queryLimit + ) + let result = try await descriptor.result(for: healthStore) + var dtos: [SampleDTO] = [] + for sample in result.addedSamples { + guard let dto = type.serialize(sample, .live) else { continue } + dtos.append(try await type.enrich?(queries, dto, sample) ?? dto) + } + if !dtos.isEmpty { + try await outbox.enqueue(typeKey: type.key, elements: dtos.map(BatchElement.sample)) + queued += dtos.count + await log.info("\(type.key): re-queued \(dtos.count) sample(s)") + } + anchor = normalize(result.newAnchor) + if result.addedSamples.count < type.queryLimit || anchor == nil { break } + } + } catch { + failed.append(type.key) + await log.error("\(type.key): \(error.localizedDescription)") + } + } + for source in AppSettings.currentEnabledSnapshots() where keys?.contains(source.key) ?? true { + do { + // Re-sends each record's current version (same uuid); a record + // that changed since the last sync goes out with the next one. + let state = snapshotCache[source.key] + ?? anchors.loadRaw(source.key).flatMap { try? JSONDecoder().decode(SnapshotState.self, from: $0) } + guard let state else { continue } + let records = try await source.read(queries, .live, nil).compactMap { record -> SampleDTO? in + guard Self.overlaps(record, start: start, end: end), + let entry = state.entries[record.uuid], + entry.fingerprint == SnapshotState.fingerprint(record) else { return nil } + return state.versioned(record, version: entry.version) + } + if !records.isEmpty { + try await outbox.enqueue(typeKey: source.key, elements: records.map(BatchElement.sample)) + queued += records.count + } + } catch { + failed.append(source.key) + await log.error("\(source.key): \(error.localizedDescription)") + } + } + await outbox.drain(force: true) + let pending = await outbox.pendingCount() + var summary = "\(reason): re-sent \(queued) record(s) from \(types.count) type(s), \(pending) batch(es) waiting" + if !failed.isEmpty { + summary += ", failed: \(failed.prefix(5).joined(separator: ", "))" + } + await status.recordSyncFinished(summary) + await status.setSyncing(false) + await log.info(summary) + return summary + } + + /// Whether a record's interval touches [start, end]; open ends match all. + static func overlaps(_ record: SampleDTO, start: Date?, end: Date?) -> Bool { + let formatter = ISO8601DateFormatter() + guard let recordStart = formatter.date(from: record.start), let recordEnd = formatter.date(from: record.end) else { + return true + } + if let start, recordEnd < start { return false } + if let end, recordStart > end { return false } + return true + } + + /// Hints plus the key of the destination set they belong to. A set that + /// changed while the request was in flight discards the answer. + private func cachedLatestHints() async -> (hints: [String: String], key: String)? { + // One snapshot of the destination list gives both the key and the + // servers asked, so the key always names the set that answered. + let snapshot = AppSettings.receivingDestinations() + let destinations = AppSettings.catchUpKey(snapshot) + if let cached = latestFetch, cached.destinations == destinations, + Date().timeIntervalSince(cached.at) < Self.latestCacheInterval { + return cached.hints.map { ($0, destinations) } + } + let hints = await latestHints(snapshot) + guard AppSettings.catchUpKey(AppSettings.receivingDestinations()) == destinations else { return nil } + latestFetch = (Date(), destinations, hints) + return hints.map { ($0, destinations) } + } + + private func sync( + _ type: SyncedType, anchor resolvedAnchor: HKQueryAnchor?, cutoffs: [String: Date], catchUpKey: String? + ) async throws -> Int { var anchor = resolvedAnchor // Date-bounding a nil-anchor query skips the historical read while // still returning a store-wide anchor, so later incremental syncs @@ -228,7 +380,13 @@ actor SyncEngine { if added.isEmpty && deleted.isEmpty { // Nothing new — the anchor covers no undelivered samples, so - // persisting it immediately is safe. + // persisting it immediately is safe, unless it ends a + // date-bounded first read whose destination set has changed. + if datePredicate != nil, let catchUpKey, + AppSettings.catchUpKey(AppSettings.receivingDestinations()) != catchUpKey { + latestFetch = nil + return queued + } if let newAnchor { anchorCache[type.key] = newAnchor anchors.save(newAnchor, for: type.key) @@ -252,9 +410,19 @@ actor SyncEngine { if !deleted.isEmpty { elements.append(.tombstone(deleted)) } - let anchorData = newAnchor.flatMap { anchors.archive($0) } - try await outbox.enqueue(typeKey: type.key, elements: elements, anchor: anchorData) + do { + // A date-bounded first read is only right for the destinations + // the hints came from. + try await outbox.enqueue( + typeKey: type.key, elements: elements, catchUpKey: datePredicate == nil ? nil : catchUpKey + ) + } catch OutboxHub.EnqueueError.destinationsChanged { + latestFetch = nil + await log.info("\(type.key): destinations changed during catch-up — will re-check on the next pass") + return queued + } if let newAnchor { + anchors.save(newAnchor, for: type.key) anchorCache[type.key] = newAnchor anchor = newAnchor } @@ -324,7 +492,7 @@ actor SyncEngine { await log.error("\(key): unreadable snapshot journal discarded") return } - try await outbox.enqueue(typeKey: key, elements: journal.elements, anchor: nil) + try await outbox.enqueue(typeKey: key, elements: journal.elements) try anchors.writeRaw(JSONEncoder().encode(journal.state), for: key) try FileManager.default.removeItem(at: url) snapshotCache[key] = journal.state diff --git a/Pulso/Sync/TriggerHub.swift b/Pulso/Sync/TriggerHub.swift index dda1af3..78f58e1 100644 --- a/Pulso/Sync/TriggerHub.swift +++ b/Pulso/Sync/TriggerHub.swift @@ -11,6 +11,7 @@ final class TriggerHub { /// Fixed rather than derived from the bundle id: Info.plist lists it /// verbatim, and a re-signed build may run under a different bundle id. static let refreshTaskID = "pulso.refresh" + static let processingTaskID = "pulso.processing" private let healthStore: HKHealthStore private let engine: SyncEngine @@ -26,7 +27,47 @@ final class TriggerHub { // MARK: - BGAppRefreshTask /// Must run before application(_:didFinishLaunchingWithOptions:) returns. - func registerBackgroundRefresh() { + func registerBackgroundTasks() { + registerBackgroundRefresh() + BGTaskScheduler.shared.register(forTaskWithIdentifier: Self.processingTaskID, using: nil) { [weak self] task in + guard let processing = task as? BGProcessingTask else { + task.setTaskCompleted(success: false) + return + } + Task { @MainActor [weak self] in + self?.handleProcessing(processing) + } + } + } + + // MARK: - BGProcessingTask + + /// A longer background window (typically overnight, on power) for + /// catching up and draining queues that a refresh couldn't finish. + private func handleProcessing(_ task: BGProcessingTask) { + scheduleProcessing() + let work = Task { + await log.info("background processing fired") + await engine.requestSync(.processing) + if !Task.isCancelled { + task.setTaskCompleted(success: true) + } + } + task.expirationHandler = { + work.cancel() + task.setTaskCompleted(success: false) + } + } + + func scheduleProcessing() { + let request = BGProcessingTaskRequest(identifier: Self.processingTaskID) + request.requiresNetworkConnectivity = true + request.requiresExternalPower = false + request.earliestBeginDate = Date(timeIntervalSinceNow: 2 * 3600) + try? BGTaskScheduler.shared.submit(request) // unsupported on simulator; fine + } + + private func registerBackgroundRefresh() { BGTaskScheduler.shared.register(forTaskWithIdentifier: Self.refreshTaskID, using: nil) { [weak self] task in guard let refresh = task as? BGAppRefreshTask else { task.setTaskCompleted(success: false) diff --git a/Pulso/Sync/Uploader.swift b/Pulso/Sync/Uploader.swift index 4793698..8bc79ed 100644 --- a/Pulso/Sync/Uploader.swift +++ b/Pulso/Sync/Uploader.swift @@ -34,7 +34,7 @@ struct HTTPUploader: Uploading { request.setValue("Bearer \(token)", forHTTPHeaderField: "Authorization") } let started = Date() - let (data, response) = try await session.upload(for: request, from: body) + let (data, response) = try await session.upload(for: request, from: body, delegate: NoRedirects.shared) return UploadResult( status: (response as? HTTPURLResponse)?.statusCode ?? 0, bodySnippet: String(decoding: data.prefix(300), as: UTF8.self), @@ -42,17 +42,18 @@ struct HTTPUploader: Uploading { ) } - /// GET /latest — per-type newest sample timestamp on the server. + /// GET <…>/latest — per-type newest sample timestamp on the server. /// nil on any failure (older servers without the endpoint included); /// callers fall back to a full backfill. - func fetchLatest(base: URL, token: String?) async -> [String: String]? { - guard let url = URL(string: base.absoluteString + "/latest") else { return nil } + func fetchLatest(url: URL, token: String?) async -> [String: String]? { var request = URLRequest(url: url) request.timeoutInterval = 10 if let token, !token.isEmpty { request.setValue("Bearer \(token)", forHTTPHeaderField: "Authorization") } - guard let (data, response) = try? await session.data(for: request), + // The request carries the token: never let a redirect take it elsewhere. + // A 3xx is non-200, so the caller falls back to a full backfill. + guard let (data, response) = try? await session.data(for: request, delegate: NoRedirects.shared), (response as? HTTPURLResponse)?.statusCode == 200 else { return nil } return try? JSONDecoder().decode([String: String].self, from: data) } @@ -62,11 +63,17 @@ struct HTTPUploader: Uploading { var request = URLRequest(url: url) request.timeoutInterval = 10 do { - let (data, response) = try await session.data(for: request) - let status = (response as? HTTPURLResponse)?.statusCode ?? 0 + // Same redirect rule as deliveries, so Test can't pass a URL + // that every upload would fail on. + let (data, response) = try await session.data(for: request, delegate: NoRedirects.shared) + let http = response as? HTTPURLResponse + let status = http?.statusCode ?? 0 if (200 ..< 300).contains(status) { return "OK — HTTP \(status)" } + if (300 ..< 400).contains(status), let location = http?.value(forHTTPHeaderField: "Location") { + return "HTTP \(status): the server redirects to \(location) — use that address instead" + } return "HTTP \(status): \(String(decoding: data.prefix(120), as: UTF8.self))" } catch { return "Unreachable: \(error.localizedDescription)" diff --git a/Pulso/UI/DestinationsView.swift b/Pulso/UI/DestinationsView.swift new file mode 100644 index 0000000..7982b64 --- /dev/null +++ b/Pulso/UI/DestinationsView.swift @@ -0,0 +1,252 @@ +import SwiftUI +import UniformTypeIdentifiers + +/// Settings › Destinations: every place data goes, each with its own +/// queue and last response. +struct DestinationsSection: View { + let services: AppServices + + var body: some View { + @Bindable var settings = services.settings + Section { + ForEach($settings.destinations) { $destination in + NavigationLink { + DestinationEditor(services: services, destination: $destination) + } label: { + VStack(alignment: .leading, spacing: 2) { + HStack { + Text(destination.name) + if !destination.enabled { + Text("off").font(.caption).foregroundStyle(.secondary) + } + } + Text("\(destination.kind.title) · \(destination.summary)") + .font(.caption) + .foregroundStyle(.secondary) + .lineLimit(1) + } + } + } + .onDelete { offsets in + // Resolve first: each removal shifts the indices after it. + let doomed = offsets.map { services.settings.destinations[$0] } + for destination in doomed { + services.removeDestination(destination) + } + } + Menu("Add Destination") { + ForEach(Destination.Kind.allCases, id: \.self) { kind in + Button(kind.title) { + services.settings.destinations.append(Destination(name: kind.title, kind: kind)) + } + } + } + } header: { + Text("Destinations") + } footer: { + Text("Every enabled destination gets every batch, from its own queue: one being down never holds up the others. A destination that is off, or added later, doesn't get what was synced before; run the Backfill shortcut to send it history.") + } + } +} + +struct DestinationEditor: View { + let services: AppServices + @Binding var destination: Destination + + @State private var bearer = "" + @State private var hmac = "" + @State private var mqttPassword = "" + @State private var testResult: String? + @State private var testing = false + @State private var pickingFolder = false + + var body: some View { + Form { + Section { + TextField("Name", text: $destination.name) + Toggle("Enabled", isOn: $destination.enabled) + LabeledContent("Kind", value: destination.kind.title) + } + + switch destination.kind { + case .rest, .pulso, .homeAssistant: + httpSections + case .mqtt: + mqttSection + case .file: + fileSection + } + + Section { + Button { + testing = true + testResult = nil + Task { + testResult = await services.testDestination(destination) + testing = false + } + } label: { + HStack { + Text("Test") + if testing { Spacer(); ProgressView() } + } + } + .disabled(testing || !destination.isConfigured) + if let testResult { + Text(testResult) + .font(.footnote) + .foregroundStyle(testResult.hasPrefix("OK") ? .green : .red) + } + if let state = services.status.destinationStatus[destination.id] { + LabeledContent("Queued batches", value: "\(state.pending)") + LabeledContent("Delivered", value: "\(state.delivered)") + if let response = state.lastResponse { + LabeledContent("Last response") { + Text(response).font(.caption).lineLimit(3).multilineTextAlignment(.trailing) + } + } + } + } + } + .navigationTitle(destination.name) + .onAppear { + bearer = destination.bearerToken ?? "" + hmac = destination.hmacSecret ?? "" + mqttPassword = destination.mqttPassword ?? "" + } + .onChange(of: destination) { _, _ in services.destinationsChanged() } + .fileImporter(isPresented: $pickingFolder, allowedContentTypes: [.folder]) { result in + let url: URL + switch result { + case .success(let picked): + url = picked + case .failure(let error): + testResult = "Couldn't open that folder: \(error.localizedDescription)" + return + } + let scoped = url.startAccessingSecurityScopedResource() + defer { if scoped { url.stopAccessingSecurityScopedResource() } } + do { + destination.folderBookmark = try url.bookmarkData() + destination.folderName = url.lastPathComponent + } catch { + testResult = "Couldn't keep access to that folder: \(error.localizedDescription)" + } + } + } + + @ViewBuilder + private var httpSections: some View { + Section { + TextField(urlPlaceholder, text: $destination.url) + .keyboardType(.URL) + .textInputAutocapitalization(.never) + .autocorrectionDisabled() + if destination.kind == .rest { + Picker("Body", selection: $destination.format) { + Text("pulso.v2 envelope").tag(Destination.BodyFormat.envelope) + Text("Bare array (Pulso v1)").tag(Destination.BodyFormat.array) + Text("NDJSON").tag(Destination.BodyFormat.ndjson) + } + Toggle("Gzip body", isOn: $destination.gzip) + Toggle("Ask /latest on first sync", isOn: $destination.catchUp) + } + } header: { + Text(destination.kind == .homeAssistant ? "Webhook" : "Endpoint") + } footer: { + Text(urlFooter) + } + + if destination.kind != .homeAssistant { + Section { + SecureField("Bearer token", text: $bearer) + .onChange(of: bearer) { _, value in destination.bearerToken = value } + SecureField("HMAC-SHA256 secret", text: $hmac) + .onChange(of: hmac) { _, value in destination.hmacSecret = value } + } header: { + Text("Authentication") + } footer: { + Text("Both optional. With a secret, each request carries X-Signature: sha256=. Stored in the Keychain.") + } + } + + Section { + ForEach($destination.headers) { $header in + HStack { + TextField("Header", text: $header.name) + .textInputAutocapitalization(.never) + .autocorrectionDisabled() + TextField("Value", text: $header.value) + .textInputAutocapitalization(.never) + .autocorrectionDisabled() + } + } + .onDelete { destination.headers.remove(atOffsets: $0) } + Button("Add Header") { + destination.headers.append(.init(name: "", value: "")) + } + } header: { + Text("Custom Headers") + } + } + + private var urlPlaceholder: String { + switch destination.kind { + case .pulso: "http://100.x.y.z:8787" + case .homeAssistant: "https://homeassistant.local:8123/api/webhook/" + default: "https://example.com/ingest" + } + } + + private var urlFooter: String { + switch destination.kind { + case .pulso: + "Base URL of a server speaking the original Pulso protocol: batches go to /ingest." + case .homeAssistant: + "Each batch triggers the webhook with a pulso.v2 envelope (uncompressed, at most 200 items, long series replaced by their length)." + default: + "Each batch is POSTed here." + } + } + + private var mqttSection: some View { + Section { + TextField("Broker host", text: $destination.mqttHost) + .textInputAutocapitalization(.never) + .autocorrectionDisabled() + TextField("Port", value: $destination.mqttPort, format: .number.grouping(.never)) + .keyboardType(.numberPad) + Toggle("TLS", isOn: $destination.mqttTLS) + TextField("Username", text: $destination.mqttUsername) + .textInputAutocapitalization(.never) + .autocorrectionDisabled() + SecureField("Password", text: $mqttPassword) + .onChange(of: mqttPassword) { _, value in destination.mqttPassword = value } + TextField("Topic prefix", text: $destination.mqttTopicPrefix) + .textInputAutocapitalization(.never) + .autocorrectionDisabled() + Toggle("Retain messages", isOn: $destination.mqttRetain) + } header: { + Text("Broker") + } footer: { + Text("QoS 1. Each batch publishes one JSON array per type to /; deletions go to /deleted.") + } + } + + private var fileSection: some View { + Section { + LabeledContent("Folder", value: destination.folderName ?? "Pulso Export (On My iPhone)") + Button("Choose Folder…") { pickingFolder = true } + if destination.folderBookmark != nil { + Button("Use the App's Own Folder", role: .destructive) { + destination.folderBookmark = nil + destination.folderName = nil + } + } + } header: { + Text("Folder") + } footer: { + Text("Appends one JSON object per line to pulso-.ndjson. Pick a folder in iCloud Drive to have it on your Mac.") + } + } +} diff --git a/Pulso/UI/SettingsView.swift b/Pulso/UI/SettingsView.swift index fa2129a..d99e6dd 100644 --- a/Pulso/UI/SettingsView.swift +++ b/Pulso/UI/SettingsView.swift @@ -3,44 +3,12 @@ import SwiftUI struct SettingsView: View { let services: AppServices - @State private var testResult: String? - @State private var testing = false @State private var confirmReimport = false var body: some View { - @Bindable var settings = services.settings NavigationStack { Form { - Section { - TextField("http://100.x.y.z:8787", text: $settings.endpoint) - .keyboardType(.URL) - .textInputAutocapitalization(.never) - .autocorrectionDisabled() - SecureField("Bearer token (optional)", text: $settings.token) - Button { - testing = true - testResult = nil - Task { - testResult = await services.testConnection() - testing = false - } - } label: { - HStack { - Text("Test Connection") - if testing { Spacer(); ProgressView() } - } - } - .disabled(testing) - if let testResult { - Text(testResult) - .font(.footnote) - .foregroundStyle(testResult.hasPrefix("OK") ? .green : .red) - } - } header: { - Text("Server") - } footer: { - Text("Base URL of your Pulso server — every sample goes there and nowhere else. Plain HTTP is fine on a Tailscale network; the tunnel is already encrypted.") - } + DestinationsSection(services: services) Section { ForEach(TypeRegistry.perObjectTypes, id: \.key) { entry in @@ -104,7 +72,7 @@ struct SettingsView: View { } header: { Text("About") } footer: { - Text("Pulso reads the Apple Health data you select and delivers it to the server you configure. No cloud, no accounts, no analytics.") + Text("Pulso reads the Apple Health data you select and delivers it to the destinations you configure. No cloud, no accounts, no analytics.") } } .navigationTitle("Settings") diff --git a/Pulso/UI/StatusView.swift b/Pulso/UI/StatusView.swift index 1e98158..f2172fa 100644 --- a/Pulso/UI/StatusView.swift +++ b/Pulso/UI/StatusView.swift @@ -21,9 +21,9 @@ struct StatusView: View { } } - if services.settings.baseURL == nil { + if !services.settings.destinations.contains(where: { $0.enabled && $0.isConfigured }) { Section { - Label("Set your server address in Settings", systemImage: "exclamationmark.triangle") + Label("Add a destination in Settings", systemImage: "exclamationmark.triangle") .foregroundStyle(.orange) } } @@ -68,18 +68,33 @@ struct StatusView: View { } Section("Delivery") { - LabeledContent("Pending batches", value: "\(status.outboxPending)") - if let retry = status.outboxNextRetry, retry > Date() { - LabeledContent("Next retry") { - Text(retry, style: .relative) + if let finished = status.lastSyncFinished { + LabeledContent("Last sync") { + Text(finished, style: .relative) } } - if let response = status.lastResponse { - LabeledContent("Last server response") { - Text(response) + if let summary = status.lastSyncSummary { + Text(summary).font(.caption).foregroundStyle(.secondary) + } + ForEach(services.settings.destinations.filter(\.enabled)) { destination in + let state = status.destinationStatus[destination.id] ?? DestinationStatus() + VStack(alignment: .leading, spacing: 2) { + HStack { + Text(destination.name) + Spacer() + Text("\(state.pending) queued").foregroundStyle(state.pending > 0 ? .orange : .secondary) + } + if let retry = state.nextRetry, retry > Date() { + HStack(spacing: 4) { + Text("retry") + Text(retry, style: .relative) + } .font(.caption) - .lineLimit(2) - .multilineTextAlignment(.trailing) + .foregroundStyle(.secondary) + } + if let response = state.lastResponse { + Text(response).font(.caption).foregroundStyle(.secondary).lineLimit(2) + } } } } diff --git a/PulsoTests/DestinationTests.swift b/PulsoTests/DestinationTests.swift new file mode 100644 index 0000000..c4c5610 --- /dev/null +++ b/PulsoTests/DestinationTests.swift @@ -0,0 +1,239 @@ +import XCTest +@testable import Pulso + +/// Scriptable transport: fails while `failing` is set, records batches. +final class MockTransport: Transport, @unchecked Sendable { + private let lock = NSLock() + private var delivered: [Batch] = [] + var failing = false + + var batches: [Batch] { + lock.lock() + defer { lock.unlock() } + return delivered + } + + func deliver(_ batch: Batch) async throws -> UploadResult { + lock.lock() + defer { lock.unlock() } + if failing { throw URLError(.cannotConnectToHost) } + delivered.append(batch) + return UploadResult(status: 200, bodySnippet: "ok", duration: 0) + } +} + +@MainActor +final class DestinationTests: XCTestCase { + private var base: URL! + + override func setUp() async throws { + base = FileManager.default.temporaryDirectory.appendingPathComponent("destination-tests-\(UUID().uuidString)") + try FileManager.default.createDirectory(at: base, withIntermediateDirectories: true) + } + + override func tearDown() async throws { + try? FileManager.default.removeItem(at: base) + } + + private func sample(_ id: String) -> BatchElement { + .sample(SampleDTO(uuid: id, type: "heartRate", start: "2026-07-06T01:12:00+01:00", + end: "2026-07-06T01:12:00+01:00", value: .number(60), unit: "count/min", source: "test")) + } + + private func pendingFiles(_ id: UUID) -> [String] { + let folder = base.appendingPathComponent(id.uuidString) + return ((try? FileManager.default.contentsOfDirectory(atPath: folder.path)) ?? []).filter { $0.hasSuffix(".json") } + } + + /// Each destination has its own queue: one that is down keeps its + /// batch while the other delivers. + func testHubFansOutAndAFailingDestinationOnlyDelaysItself() async throws { + var up = Destination(name: "up", kind: .file) + up.enabled = true + var down = Destination(name: "down", kind: .file) + down.enabled = true + let upTransport = MockTransport() + let downTransport = MockTransport() + downTransport.failing = true + let transports: [UUID: MockTransport] = [up.id: upTransport, down.id: downTransport] + let destinations = [up, down] + let hub = OutboxHub( + root: base, log: LogStore(directory: base), status: AppStatus(), + destinations: { destinations }, + transport: { destination in transports[destination.id] } + ) + + try await hub.enqueue(typeKey: "heartRate", elements: [sample("a"), sample("b")]) + await hub.drain(force: true) + + XCTAssertEqual(upTransport.batches.count, 1) + XCTAssertEqual(upTransport.batches.first?.elements.count, 2) + XCTAssertEqual(pendingFiles(up.id), []) + XCTAssertEqual(pendingFiles(down.id).count, 1, "undelivered batch must stay on disk") + + downTransport.failing = false + await hub.drain(force: true) + XCTAssertEqual(downTransport.batches.count, 1) + XCTAssertEqual(pendingFiles(down.id), []) + } + + func testHubRefusesToEnqueueWithoutADestination() async { + let hub = OutboxHub(root: base, log: LogStore(directory: base), status: AppStatus(), destinations: { [] }) + do { + try await hub.enqueue(typeKey: "heartRate", elements: [sample("a")]) + XCTFail("enqueue must throw so the caller keeps its anchor") + } catch {} + } + + /// RFC 4231 test case 2. + func testHMACMatchesTheKnownVector() { + let mac = HTTPTransport.hmacHex(Data("what do ya want for nothing?".utf8), secret: "Jefe") + XCTAssertEqual(mac, "5bdcc146bf60754e6a042426089575c75a003f089d2739839dec58b964ec3843") + } + + func testEnvelopeCarriesSchemaDeviceAndItems() throws { + let batch = Batch(id: UUID(), typeKey: "heartRate", createdAt: Date(), anchor: nil, + elements: [sample("a"), .tombstone(["gone"])]) + let device = DeviceInfo(id: "D", name: "Phone", model: "iPhone", systemVersion: "26.0", + appVersion: "1.0 (1)", bundleId: "x") + let body = try BatchBody.make(batch, format: .envelope, device: device) + let json = try XCTUnwrap(JSONSerialization.jsonObject(with: body) as? [String: Any]) + XCTAssertEqual(json["schema"] as? String, "pulso.v2") + XCTAssertEqual(json["type"] as? String, "heartRate") + XCTAssertEqual((json["device"] as? [String: Any])?["id"] as? String, "D") + let items = try XCTUnwrap(json["items"] as? [[String: Any]]) + XCTAssertEqual(items.count, 2) + XCTAssertEqual(items[0]["uuid"] as? String, "a") + XCTAssertEqual(items[1]["deleted"] as? [String], ["gone"]) + + let array = try BatchBody.make(batch, format: .array, device: device) + XCTAssertEqual(try JSONDecoder().decode([BatchElement].self, from: array), batch.elements) + } + + func testSlimmingReplacesLongSeriesWithTheirLength() { + var dto = SampleDTO(uuid: "r", type: "workoutRoute", start: "s", end: "e", source: "x") + dto.details = ["locations": .array(Array(repeating: .number(1), count: 500)), "workoutUUID": "w"] + guard case .sample(let slim) = BatchBody.slimmed(.sample(dto)) else { return XCTFail() } + XCTAssertNil(slim.details?["locations"]) + XCTAssertEqual(slim.details?["locationsCount"], .number(500)) + XCTAssertEqual(slim.details?["workoutUUID"], "w") + } + + func testPulsoKindPostsToIngestUnderTheBaseURL() { + var destination = Destination(name: "p", kind: .pulso) + destination.url = "100.64.0.7:8787/" + XCTAssertEqual(destination.ingestURL?.absoluteString, "http://100.64.0.7:8787/ingest") + XCTAssertEqual(destination.format, .array) + XCTAssertEqual(destination.latestURL?.absoluteString, "http://100.64.0.7:8787/latest") + var rest = Destination(name: "r", kind: .rest) + rest.url = "health.example.net:8797/ingest/pulso" + XCTAssertEqual(rest.ingestURL?.absoluteString, "https://health.example.net:8797/ingest/pulso") + XCTAssertNil(rest.latestURL, "catch-up is opt-in for REST") + } + + /// The anchor is shared, so the catch-up point must suit every + /// destination: earliest timestamp per type, and only types all have. + func testCatchUpUsesTheEarliestAnswerAndOnlySharedTypes() { + let merged = CatchUp.earliest([ + ["heartRate": "2026-07-15T08:59:12+01:00", "stepCount": "2026-07-15T09:00:00+01:00"], + ["heartRate": "2026-07-10T08:00:00+02:00"], + ]) + XCTAssertEqual(merged, ["heartRate": "2026-07-10T08:00:00+02:00"]) + XCTAssertEqual(CatchUp.earliest([]), [:]) + XCTAssertEqual( + CatchUp.earliest([["heartRate": "2026-07-15T08:59:12+01:00"], ["heartRate": ""]]), [:], + "an unparseable answer must not leave the other server's cutoff in place" + ) + } + + func testResendRangeMatchesOverlappingRecordsOnly() { + let day = SampleDTO(uuid: "d", type: "activitySummary", start: "2026-10-21T00:00:00+02:00", + end: "2026-10-22T00:00:00+02:00", source: "x") + let iso = ISO8601DateFormatter() + XCTAssertTrue(SyncEngine.overlaps(day, start: iso.date(from: "2026-10-21T12:00:00Z"), end: nil)) + XCTAssertFalse(SyncEngine.overlaps(day, start: iso.date(from: "2026-10-23T00:00:00Z"), end: nil)) + XCTAssertFalse(SyncEngine.overlaps(day, start: nil, end: iso.date(from: "2026-10-20T00:00:00Z"))) + XCTAssertTrue(SyncEngine.overlaps(day, start: nil, end: nil)) + } + + /// A destination whose URL is being edited still gets the batch queued, + /// so it delivers later instead of missing what synced meanwhile; one + /// added and never set up gets nothing, so its queue can't grow forever. + func testUnconfiguredDestinationsQueueOnlyIfTheyWereConfiguredBefore() async throws { + let file = Destination(name: "file", kind: .file) + var editing = Destination(name: "editing", kind: .rest) + editing.wasConfigured = true + let abandoned = Destination(name: "abandoned", kind: .mqtt) + XCTAssertFalse(editing.isConfigured) + XCTAssertFalse(abandoned.isConfigured) + let destinations = [file, editing, abandoned] + // Like production: an unconfigured destination has no transport. + let hub = OutboxHub( + root: base, log: LogStore(directory: base), status: AppStatus(), + destinations: { destinations }, + transport: { $0.isConfigured ? MockTransport() : nil } + ) + try await hub.enqueue(typeKey: "heartRate", elements: [sample("a")]) + await hub.drain(force: true) + XCTAssertEqual(pendingFiles(file.id), []) + XCTAssertEqual(pendingFiles(editing.id).count, 1, "waits for its URL") + XCTAssertEqual(pendingFiles(abandoned.id), [], "never configured, never queued") + } + + /// Fields added in later versions must not make a stored list unreadable. + func testDestinationDecodesWithMissingFields() throws { + let id = UUID() + let json = #"[{"id":"\#(id.uuidString)","kind":"rest","url":"https://x.example/in"}]"# + let decoded = try JSONDecoder().decode([Destination].self, from: Data(json.utf8)) + XCTAssertEqual(decoded.first?.id, id) + XCTAssertEqual(decoded.first?.url, "https://x.example/in") + XCTAssertEqual(decoded.first?.gzip, true) + XCTAssertEqual(decoded.first?.format, .envelope) + XCTAssertEqual(decoded.first?.enabled, true) + XCTAssertEqual(decoded.first?.wasConfigured, true, "a stored destination with a URL counts as configured") + } + + /// Clearing a URL in the session the destination was added must not + /// un-set wasConfigured: the in-memory copy never saw the flag. + func testWasConfiguredSurvivesSavingAnUnconfiguredCopy() { + let defaults = UserDefaults.standard + let saved = defaults.data(forKey: SettingsKeys.destinations) + defer { defaults.set(saved, forKey: SettingsKeys.destinations) } + + var destination = Destination(name: "r", kind: .rest) + destination.url = "https://a.example/in" + AppSettings.storeDestinations([destination]) + XCTAssertEqual(AppSettings.currentDestinations().first?.wasConfigured, true) + + destination.url = "" + XCTAssertFalse(destination.wasConfigured, "the edited copy still has the original flag") + AppSettings.storeDestinations([destination]) + XCTAssertEqual(AppSettings.currentDestinations().first?.wasConfigured, true) + XCTAssertEqual(AppSettings.receivingDestinations().map(\.id), [destination.id]) + } + + /// A batch read with catch-up hints may only go to the destination set + /// the hints came from; a destination added meanwhile needs the history. + func testCatchUpBatchIsRefusedWhenTheDestinationSetChanged() async throws { + let a = Destination(name: "a", kind: .file) + let b = Destination(name: "b", kind: .file) + var destinations = [a] + let lock = NSLock() + let hub = OutboxHub( + root: base, log: LogStore(directory: base), status: AppStatus(), + destinations: { lock.lock(); defer { lock.unlock() }; return destinations }, + transport: { _ in nil } + ) + let key = AppSettings.catchUpKey([a]) + try await hub.enqueue(typeKey: "heartRate", elements: [sample("1")], catchUpKey: key) + XCTAssertEqual(pendingFiles(a.id).count, 1) + + lock.lock(); destinations = [a, b]; lock.unlock() + do { + try await hub.enqueue(typeKey: "heartRate", elements: [sample("2")], catchUpKey: key) + XCTFail("must refuse: b joined after the hints were fetched") + } catch OutboxHub.EnqueueError.destinationsChanged {} + XCTAssertEqual(pendingFiles(a.id).count, 1, "nothing written on refusal") + XCTAssertEqual(pendingFiles(b.id), []) + } +} diff --git a/PulsoTests/MQTTPacketTests.swift b/PulsoTests/MQTTPacketTests.swift new file mode 100644 index 0000000..e3f36f7 --- /dev/null +++ b/PulsoTests/MQTTPacketTests.swift @@ -0,0 +1,165 @@ +import XCTest +@testable import Pulso + +final class MQTTPacketTests: XCTestCase { + private func config(username: String? = nil, password: String? = nil) -> MQTTConfig { + MQTTConfig( + host: "broker.example", port: 1883, useTLS: false, + username: username, password: password, clientID: "pulso", keepAlive: 0x1234 + ) + } + + func testRemainingLengthBoundaries() { + let cases: [(Int, [UInt8])] = [ + (0, [0x00]), + (127, [0x7f]), + (128, [0x80, 0x01]), + (16_383, [0xff, 0x7f]), + (16_384, [0x80, 0x80, 0x01]), + (2_097_151, [0xff, 0xff, 0x7f]), + (2_097_152, [0x80, 0x80, 0x80, 0x01]), + (268_435_455, [0xff, 0xff, 0xff, 0x7f]), + ] + for (length, expected) in cases { + XCTAssertEqual(MQTTPacket.encodeRemainingLength(length), expected, "length \(length)") + } + } + + func testConnectWithoutCredentials() { + XCTAssertEqual(MQTTPacket.connect(config()), Data([ + 0x10, 0x11, + 0x00, 0x04, 0x4d, 0x51, 0x54, 0x54, 0x04, + 0x02, 0x12, 0x34, + 0x00, 0x05, 0x70, 0x75, 0x6c, 0x73, 0x6f, + ])) + } + + func testRemainingLengthValidationRejectsOverflowWithoutAllocatingPayload() throws { + try MQTTPacket.validateLength(268_435_455) + for length in [-1, 268_435_456, Int.max] { + XCTAssertThrowsError(try MQTTPacket.validateLength(length)) { error in + guard let mqttError = error as? MQTTError, case .protocolViolation = mqttError else { + return XCTFail("Expected protocolViolation, got \(error)") + } + } + } + } + + func testConnectWithCredentials() { + XCTAssertEqual(MQTTPacket.connect(config(username: "u", password: "pw")), Data([ + 0x10, 0x18, + 0x00, 0x04, 0x4d, 0x51, 0x54, 0x54, 0x04, + 0xc2, 0x12, 0x34, + 0x00, 0x05, 0x70, 0x75, 0x6c, 0x73, 0x6f, + 0x00, 0x01, 0x75, + 0x00, 0x02, 0x70, 0x77, + ])) + } + + func testConnectOmitsEmptyCredentialsAndCanDisableCleanSession() throws { + XCTAssertEqual(MQTTPacket.connect(config(username: "", password: "")), MQTTPacket.connect(config())) + let usernameOnly = try XCTUnwrap(MQTTPacket.parseFrame( + MQTTPacket.connect(config(username: "u", password: ""), cleanSession: false) + )) + XCTAssertEqual(usernameOnly.body[7], 0x80) + XCTAssertEqual(Array(usernameOnly.body.suffix(3)), [0, 1, 0x75]) + } + + func testConnectValidationRejectsPasswordWithoutUsername() throws { + let usernames: [String?] = [nil, ""] + for username in usernames { + XCTAssertThrowsError(try MQTTPacket.validateCredentials(config(username: username, password: "pw"))) { error in + guard let mqttError = error as? MQTTError, case .protocolViolation = mqttError else { + return XCTFail("Expected protocolViolation, got \(error)") + } + } + } + try MQTTPacket.validateCredentials(config()) + try MQTTPacket.validateCredentials(config(username: "", password: "")) + try MQTTPacket.validateCredentials(config(username: "u", password: "pw")) + try MQTTPacket.validateCredentials(config(username: "u")) + } + + func testPublishQoSOneHeaderAndPacketID() throws { + for retain in [false, true] { + let packet = MQTTPacket.publish(topic: "a/b", payload: Data([0, 0xff]), packetID: 0x1234, retain: retain) + XCTAssertEqual(packet, Data([ + retain ? 0x33 : 0x32, 0x09, + 0x00, 0x03, 0x61, 0x2f, 0x62, 0x12, 0x34, 0x00, 0xff, + ])) + let frame = try XCTUnwrap(MQTTPacket.parseFrame(packet)) + XCTAssertEqual(frame.type, 3) + XCTAssertEqual(frame.flags, retain ? 3 : 2) + } + } + + func testPublishUsesUTF8ByteLength() throws { + let packet = MQTTPacket.publish(topic: "é/心", payload: Data(), packetID: 1, retain: false) + let frame = try XCTUnwrap(MQTTPacket.parseFrame(packet)) + XCTAssertEqual(frame.body, Data([0, 6, 0xc3, 0xa9, 0x2f, 0xe5, 0xbf, 0x83, 0, 1])) + } + + func testConnectUsesUTF8ByteLength() throws { + var settings = config() + settings.clientID = "心" + let frame = try XCTUnwrap(MQTTPacket.parseFrame(MQTTPacket.connect(settings))) + XCTAssertEqual(Array(frame.body.suffix(5)), [0, 3, 0xe5, 0xbf, 0x83]) + } + + func testParsePartialAndExactFrames() throws { + let connack = Data([0x20, 0x02, 0x00, 0x00]) + let puback = Data([0x40, 0x02, 0x12, 0x34]) + for packet in [connack, puback] { + for count in 0 ..< packet.count { + XCTAssertNil(MQTTPacket.parseFrame(Data(packet.prefix(count)))) + } + let frame = try XCTUnwrap(MQTTPacket.parseFrame(packet)) + XCTAssertEqual(frame.type, packet[0] >> 4) + XCTAssertEqual(frame.flags, 0) + XCTAssertEqual(frame.body, Data(packet.suffix(2))) + XCTAssertEqual(frame.consumed, 4) + } + } + + func testParseConcatenatedFramesAndNonzeroBufferIndices() throws { + var buffer = Data([0x20, 0x02, 0x00, 0x00, 0x40, 0x02, 0xab, 0xcd]) + let first = try XCTUnwrap(MQTTPacket.parseFrame(buffer)) + XCTAssertEqual(first.type, 2) + XCTAssertEqual(first.body, Data([0, 0])) + XCTAssertEqual(first.consumed, 4) + buffer.removeFirst(first.consumed) + let second = try XCTUnwrap(MQTTPacket.parseFrame(buffer)) + XCTAssertEqual(second.type, 4) + XCTAssertEqual(second.body, Data([0xab, 0xcd])) + XCTAssertEqual(second.consumed, 4) + buffer.removeFirst(second.consumed) + XCTAssertNil(MQTTPacket.parseFrame(buffer)) + } + + func testParsePartialMultibyteLengthAndLargePayload() throws { + let payload = Data(repeating: 0xa5, count: 2_097_152) + let packet = MQTTPacket.publish(topic: "x", payload: payload, packetID: .max, retain: false) + for count in 1 ... 5 { + XCTAssertNil(MQTTPacket.parseFrame(Data(packet.prefix(count)))) + } + XCTAssertNil(MQTTPacket.parseFrame(Data(packet.dropLast()))) + let frame = try XCTUnwrap(MQTTPacket.parseFrame(packet)) + XCTAssertEqual(frame.consumed, packet.count) + XCTAssertEqual(Array(frame.body.prefix(5)), [0, 1, 0x78, 0xff, 0xff]) + XCTAssertEqual(Data(frame.body.dropFirst(5)), payload) + } + + func testParseRejectsMalformedRemainingLengths() { + XCTAssertNil(MQTTPacket.parseFrame(Data([0x20, 0xff, 0xff, 0xff, 0xff]))) + XCTAssertNil(MQTTPacket.parseFrame(Data([0x20, 0x80, 0x00]))) + } + + func testControlPacketsHaveEmptyBodies() throws { + XCTAssertEqual(MQTTPacket.disconnect(), Data([0xe0, 0])) + XCTAssertEqual(MQTTPacket.pingreq(), Data([0xc0, 0])) + let frame = try XCTUnwrap(MQTTPacket.parseFrame(MQTTPacket.pingreq())) + XCTAssertEqual(frame.type, 12) + XCTAssertTrue(frame.body.isEmpty) + XCTAssertEqual(frame.consumed, 2) + } +} diff --git a/README.md b/README.md index 830142d..4212a8c 100644 --- a/README.md +++ b/README.md @@ -6,6 +6,16 @@

Continuously deliver your Apple Health data to a server you own.

+> **This fork** ([Programcode/pulso](https://github.com/Programcode/pulso), from +> [lenilsonjr/pulso](https://github.com/lenilsonjr/pulso)) reads every +> HealthKit type the iOS SDK lets a third-party app read — the type list is +> generated from the SDK headers at build time — and sends it to several +> destinations at once: HTTPS/webhooks (gzip, custom headers, bearer token, +> HMAC signature), the original Pulso protocol, Home Assistant, MQTT, and +> NDJSON files. It adds Shortcuts actions and ships as an unsigned IPA built +> by GitHub Actions. The wire format is documented in +> [`docs/PROTOCOL.md`](docs/PROTOCOL.md) (v2 section). + Pulso is an open-source iOS app with one job: read the Apple Health data you select and ship it — incrementally, in the background, surviving crashes and outages — to an HTTP endpoint you configure. Self-hosted personal telemetry. @@ -20,13 +30,12 @@ outages — to an HTTP endpoint you configure. Self-hosted personal telemetry. - All data flows **device → your server**. There is no cloud, no account, no analytics, no telemetry, and no third-party code (zero dependencies). -- The app makes exactly **three kinds of network calls**, all to the server - you configure: `POST /ingest` (your health samples), `GET /health` (the - "Test Connection" button), and `GET /latest` (asks your server what it - already holds, so a reinstall doesn't re-send all history). Nothing else, - ever. +- The app talks only to the destinations you configure: it POSTs batches + (or publishes them over MQTT, or writes files), sends a test request when + you tap Test, and, if a destination has catch-up enabled, asks its + `/latest` what it already holds. Nothing else, ever. - Pending data on the device is excluded from device backups. -- The optional bearer token is stored in the iOS Keychain. +- Bearer tokens, HMAC secrets and MQTT passwords are stored in the iOS Keychain. ## Why it exists @@ -40,20 +49,25 @@ continuously, a dedicated app is the only honest path. This is that app. ## How it works ``` -HealthKit ──anchored queries──► SyncEngine ──batches──► Outbox (disk) ──HTTP──► your server - ▲ │ - observer / BGTask / foreground triggers retry w/ backoff + ┌─► queue (disk) ──HTTPS──► webhook / Pulso server / Home Assistant +HealthKit ──anchored queries──► SyncEngine ──batches──┼─► queue (disk) ──MQTT───► broker + ▲ │ │ + observer / BGTasks / Shortcuts / app open └─► queue (disk) ──file───► NDJSON folder + each queue retries on its own, with backoff ``` - **Anchored queries** (`HKAnchoredObjectQuery`) are the sync cursor: each pass returns exactly the new samples and deletions since the last anchor — rewrites and late backfills from other apps included. -- **The outbox** writes every batch to disk *before* the first upload - attempt and deletes it only on a server 2xx. **The anchor advances only - when the server ACKs.** Crash anywhere and the worst case is a re-send, - which the server dedupes by sample uuid. -- **Triggers**: HealthKit background delivery (immediate for sleep and - workouts, hourly for the rest), an opportunistic `BGAppRefreshTask`, and a +- **The queues** (one per destination that receives batches) get every + batch on disk *before* the first delivery attempt, and each deletes it + only once its own destination accepts it (a 2xx, an MQTT PUBACK, a + completed file write). **The anchor advances once the batch is in every + queue**, so a destination that is down delays only itself. Crash anywhere + and the worst case is a re-send, which receivers dedupe by sample uuid. +- **Triggers**: HealthKit background delivery (immediate for categories, + workouts and other events, hourly for quantity types), an opportunistic + `BGAppRefreshTask`, a `BGProcessingTask`, the Shortcuts actions, and a full sync every time you open the app. **Honest expectations:** iOS does not run daemons, and health data is @@ -63,15 +77,60 @@ ceiling on iOS, and it's good enough. ### Synced types -58 types across nine groups: sleep & circadian (sleep stages, daylight -time, wrist temperature, breathing disturbances, stand hours, mindfulness), -workouts (with full metadata and heart-rate statistics), energy & activity, -cardio & recovery, body composition, running & gait form, audio exposure, -nutrition, and lifestyle (sexual activity). The complete list with units lives in -[`docs/PROTOCOL.md`](docs/PROTOCOL.md). The architecture is type-generic — -adding a type is a row in -[`TypeRegistry.swift`](Pulso/Health/TypeRegistry.swift). Every type can be -toggled individually in Settings. +Everything the iOS SDK declares that a third-party app can read, on the iOS +the phone runs (iOS 17+): about 120 quantity types, 70 category types +(symptoms, cycle tracking, sleep stages, events…), workouts (with events, +multisport activities, all statistics and links to their routes and +heart-rate samples), workout routes with GPS points, heartbeat series, ECGs +with voltages, audiograms, state of mind and GAD-7/PHQ-9 (iOS 18), vision +prescriptions, medication doses and the medication list (iOS 26), blood +pressure and food correlations, characteristics (birth date, sex, blood +type, skin type, wheelchair use, move mode) and daily activity rings. + +[`scripts/gen_healthkit_catalog.py`](scripts/gen_healthkit_catalog.py) +generates [`HealthKitCatalog.generated.swift`](Pulso/Health/HealthKitCatalog.generated.swift) +from the SDK's HealthKit headers; CI regenerates it before every IPA build, +so a newer Xcode brings its new types along. Settings → Coverage on the +phone lists what this build syncs and what it skips, with the reason. + +Not synced: clinical records (FHIR) and CDA documents. Both need the +`health-records` entitlement plus per-record approval, which re-signing +certificates generally can't grant; HealthKit itself withholds a few +identifiers (shown in Coverage). + +Every type can be toggled in Settings (all on by default). Vision +prescriptions and medications are shared item by item: tap their buttons +under Settings → Health Access. + +### Destinations + +Settings → Destinations. Several can be enabled; each has its own durable +queue and retry backoff, and a batch is deleted from a queue only after +that destination accepted it. + +- **HTTPS / webhook** — POSTs the pulso.v2 envelope (or a bare array, or + NDJSON), gzip optional, custom headers, `Authorization: Bearer`, and with + an HMAC secret `X-Signature: sha256=` over the body as sent. +- **Pulso server** — the original protocol, for the Rust reference server. +- **Home Assistant webhook** — the envelope, small batches, long series + reduced to their length. +- **MQTT** — QoS 1, one JSON message per type per batch on `/`. +- **File export** — NDJSON appended to `pulso-.ndjson` in the app's + folder (Files → On My iPhone → Pulso) or a folder you pick, e.g. in + iCloud Drive. + +### Shortcuts + +- **Sync Health Data** — optional types; with a date range it re-sends that + period. +- **Backfill Health Data** — re-sends history (optionally since a date) for + some or all types; opens the app so it can run long. +- **Get Last Sync Status** — returns a text summary: last sync, what was + sent, what is queued per destination. + +Run them from Personal Automations at fixed times. HealthKit is only +readable while the phone is unlocked, so a sync that fires while it is +locked does nothing until the next one. ## Quick start @@ -95,10 +154,20 @@ docker run -p 8787:8787 -v pulso-data:/data pulso-server Your data ends up in `data/.ndjson` — one JSON object per line, inspectable with `cat`. -### 2. Build the app +### 2. Install the app + +**Unsigned IPA (this fork):** every push to `main` builds +`Pulso---unsigned.ipa` in GitHub Actions (artifact +*Pulso-unsigned-ipa*); a `v*` tag also attaches it to a GitHub Release. +Re-sign it with your own certificate (Signulous, AltStore, …). The binary +carries the HealthKit and HealthKit background-delivery entitlements; your +provisioning profile must allow them. The bundle id is +`com.programcode.pulso` unless you set the `PULSO_BUNDLE_ID` repository +variable or pass `bundle_id` when running the workflow by hand. -Open `Pulso.xcodeproj` in Xcode 15+, select the **Pulso** target → Signing & -Capabilities, pick your team, and run on your iPhone (iOS 17+). +**From source:** open `Pulso.xcodeproj` in Xcode 26+, select the **Pulso** +target → Signing & Capabilities, pick your team, and run on your iPhone +(iOS 17+). - **Free Apple ID:** works, but apps expire after 7 days and must be re-installed from Xcode. @@ -109,8 +178,9 @@ Capabilities, pick your team, and run on your iPhone (iOS 17+). ### 3. Configure and sync -In the app: Settings → enter your server URL (e.g. `http://100.64.0.7:8787`), -optionally a token → Test Connection → Status → Grant Health Access → Sync Now. +In the app: Settings → Add Destination → enter its URL and credentials → +Test → Status → Grant Health Access → Sync Now. An existing Pulso install +keeps its server as a "Pulso server" destination. The first sync imports your full history (can be 100k+ samples — keep the app open and unlocked; a progress note shows in Status). It ships in 5,000- @@ -149,8 +219,9 @@ history the server already has (via `GET /latest`), but that check has no per-device awareness — samples that only exist on the "new" device and predate the server's newest data would otherwise be skipped. -**Workout GPS routes? ECG? Writing data back?** -Out of scope for v1. Routes (`HKWorkoutRoute`) are the most likely v2 item. +**Workout GPS routes? ECG?** +Both synced in this fork (see `details` in the v2 protocol). Writing data +back to Health is out of scope. ## Repository layout diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index 90e366d..7588cd6 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -283,3 +283,162 @@ The reference server appends each element to `data/.ndjson` (tombstones to `data/_deleted.ndjson`), one JSON object per line, adding a `receivedAt` field with the server's local time. Inspect with `cat`, `jq`, or anything that reads lines. + +--- + +# v2 (this fork): every HealthKit type, several destinations + +v2 keeps v1's sample object and adds to it; everything above still holds +for a "Pulso server" destination, which sends exactly the v1 body. + +## Request bodies + +Each destination picks one of three bodies for a batch: + +- **pulso.v2 envelope** (HTTPS / webhook default, and Home Assistant): + + ```json + { + "schema": "pulso.v2", + "batchId": "6F1C…", + "type": "heartRate", + "createdAt": "2026-10-22T08:00:03Z", + "sentAt": "2026-10-22T08:00:04Z", + "device": {"id": "", "name": "iPhone", "model": "iPhone", + "systemVersion": "26.1", "appVersion": "0.1.0 (12)", "bundleId": "com.programcode.pulso"}, + "items": [ ] + } + ``` + + A test request (the Test button) is `{"schema": "pulso.v2", "test": true, + "items": [], …}`; receivers should accept it and store nothing. + +- **Bare array**: the v1 body. +- **NDJSON**: one item per line (also what file export writes). + +`type` names the type the batch was read for; items carry their own `type` +too. Batches hold one type, except snapshot batches. + +## Headers + +- `Content-Encoding: gzip` when the destination has gzip on (default for + HTTPS; off for Home Assistant). +- `Authorization: Bearer ` when a token is set. +- `X-Signature: sha256=` when an HMAC secret is set: HMAC-SHA256 of the + request body **exactly as sent** (after gzip), keyed with the secret. +- `X-Pulso-Batch: ` and `X-Pulso-Type: ` on every batch. +- Any custom headers configured on the destination. + +Success is any 2xx. Anything else, or no response, keeps the batch queued +and retries with exponential backoff (1 min → 1 h). + +## Additional sample fields + +All optional, omitted when absent. + +| Field | Notes | +|---|---| +| `kind` | `quantity`, `category`, `workout`, `route`, `heartbeatSeries`, `electrocardiogram`, `audiogram`, `stateOfMind`, `scoredAssessment`, `visionPrescription`, `medicationDoseEvent`, `correlation`, `characteristics`, `activitySummary`, `medication` | +| `sourceVersion`, `sourceProductType`, `sourceOSVersion` | from `HKSourceRevision` | +| `device` | the `HKDevice` fields that are set: `name`, `manufacturer`, `model`, `hardwareVersion`, `firmwareVersion`, `softwareVersion`, `localIdentifier`, `udiDeviceIdentifier` | +| `metadata` | now the sample's **full** HealthKit metadata for every type (v1 sent it only for workouts). `timeZone` and `protectionUsed` are still folded in as in v1. Nested values pass through as JSON arrays/objects. | +| `details` | type-specific structure, below | + +## Type keys and units + +Every quantity, category and characteristic identifier in the iOS SDK the +app was built with is synced. The key is the identifier without its prefix, +in lower camel case (`HKQuantityTypeIdentifierVO2Max` → `vo2Max`). The list +is generated from the SDK headers by `scripts/gen_healthkit_catalog.py`; +Settings → Coverage on the phone shows what this build syncs and what it +skips, with the reason. + +Units: the 58 v1.1 types keep their v1.1 units (tables above). Every other +quantity uses the canonical unit HealthKit's headers name for it (`%` +values are still scaled to 0–100, and `count/s` is sent as `count/min`). +Category values are the Swift case names of the SDK's value enum +(`HKCategoryValueSeverity.moderate` → `"moderate"`), `value_` for ones +the build doesn't know. + +Other type keys: `workout`, `workoutRoute`, `heartbeatSeries`, +`electrocardiogram`, `audiogram`, `stateOfMind` (iOS 18+), `gad7`, `phq9` +(iOS 18+), `visionPrescription`, `medicationDoseEvent` (iOS 26+), +`bloodPressure` and `food` (correlations), and the snapshot types +`characteristics`, `activitySummary`, `medication` (iOS 26+). + +## `details` by kind + +- **quantity**: for series samples, `count` and (discrete types) `minimum`, + `maximum`, `average`, `mostRecent`, in the sample's unit. +- **workout**: `events` (`type`, `start`, `end`, `duration`, `metadata`), + `activities` (multisport segments: `uuid`, `activityType`, `locationType`, + `swimmingLocationType`, `lapLengthMeters`, `start`, `end`, `duration`, + `metadata`, `events`, `statistics`), `statistics` (per type key: `unit`, + `sum`, `average`, `minimum`, `maximum`, `mostRecent`), `routeUUIDs`, + `heartRateSampleUUIDs` and `heartRateLink` (`associated` when the recording + app tied the samples to the workout, `timeRange` when they were matched by + time). The v1 workout fields are unchanged. +- **route** (`workoutRoute`): `locationFields` names the columns of + `locations`, each a row `[secondsFromStart, latitude, longitude, altitude, + horizontalAccuracy, verticalAccuracy, speed, course, speedAccuracy, + courseAccuracy]`; `workoutUUID` when the parent workout was found. +- **heartbeatSeries**: `beats` rows `[secondsFromStart, precededByGap]`. +- **electrocardiogram**: `classification`, `symptomsStatus`, + `averageHeartRate`, `samplingFrequencyHz`, `numberOfVoltageMeasurements`, + `lead`, `voltages` rows `[secondsFromStart, microvolts]`. +- **audiogram**: `points`, each with `frequencyHz` and `tests` (`side`, + `conductionType`, `masked`, `sensitivityDBHL`, clamping range) on iOS + 18.1+, or `leftEarDBHL` / `rightEarDBHL` before. +- **stateOfMind**: `kind`, `valence`, `valenceClassification`, `labels`, + `associations`. +- **scoredAssessment** (`gad7`, `phq9`): `score` (also the `value`), `risk`, + `answers`. +- **visionPrescription**: `prescriptionType`, `dateIssued`, `expirationDate`, + `rightEye` / `leftEye` (sphere, cylinder, axis, add power, vertex and + pupillary distances, prism; contacts: base curve, diameter), `brand`. +- **medicationDoseEvent**: `logStatus`, `scheduleType`, `scheduledDate`, + `doseQuantity`, `scheduledDoseQuantity`, `unit`, `medicationConcept`, and + `medicationName` / `medicationNickname` when the medication is shared. +- **correlation**: `objects`, the member samples' `uuid`, `type`, `value`, + `unit` (the members are also sent as their own samples). + +## Snapshots + +Characteristics, daily activity summaries and the medication list have no +sync cursor, so they are re-read and sent only when their content changes. +Each version is a new record whose `uuid` is never reused: the record key, +a random epoch, and a version counter (`activitySummary-2026-10-21-3f9a0c1b-v2`). +`details.recordKey` holds the stable key, and the same batch tombstones the +version it replaces. Characteristics and medications that disappear from +Health are tombstoned too. A receiver that keeps the first copy of a uuid +and one that upserts both end up with exactly the latest version. + +After "Re-import Full History" the app starts a new epoch and does not +tombstone the versions sent before it, so consumers should treat the newest +record per `recordKey` as current (the self-hosted health-ingest receiver +does this on insert). + +- `characteristics` (key `characteristics`): `details` holds + `dateOfBirth`, `biologicalSex`, `bloodType`, `fitzpatrickSkinType`, + `wheelchairUse`, `activityMoveMode`. +- `activitySummary` (key `activitySummary-YYYY-MM-DD`, start/end = that + day): move, exercise and stand values with their goals, `activityMoveMode`, + `paused`. The first read covers all history; later reads re-check the + last 35 days, or from the last day sent if that is earlier. +- `medication` (key `medication-`): `displayText`, `nickname`, + `generalForm`, `isArchived`, `hasSchedule`, `relatedCodings`. + +## Deletions + +Tombstones are unchanged: `{"deleted": [uuids]}`. + +## Other destinations + +- **MQTT**: QoS 1. Each batch publishes one JSON array (bare-array form) per + type to `/`, and tombstones to `/deleted`. The Test + button publishes an envelope to `/test`. +- **Home Assistant**: the envelope, uncompressed, at most 200 items per + request, with arrays longer than 50 entries in `details` replaced by + `Count`. +- **File export**: NDJSON lines appended to `pulso-YYYY-MM-DD.ndjson` (the + day of export) in the chosen folder.