1 contributor
import CryptoKit
import Foundation
import HealthKit
import SQLite3
private enum SQLiteHealthArchiveStoreError: Error {
case openFailed(String)
case prepareFailed(String)
case stepFailed(String)
case incompatibleSchema(Int)
case exportEncodingFailed
}
// Interface updated 2026-05-24 — see AGENTS.md
actor SQLiteHealthArchiveStore: HealthArchiveStore {
static let shared = SQLiteHealthArchiveStore()
nonisolated static let defaultDatabaseURL = URL.applicationSupportDirectory.appending(path: "HealthProbeArchive.sqlite")
nonisolated private static let archiveSchemaVersion = 2
nonisolated private static let requiredArchiveV2Tables: [String] = [
"schema_migrations",
"archive_metadata",
"device_chains",
"observations",
"sample_types",
"observation_type_runs",
"sources",
"source_revisions",
"hk_devices",
"metadata_blobs",
"samples",
"sample_versions",
"sample_observation_events",
"sample_visibility_ranges",
"sample_relationships",
"observation_type_summaries",
"type_capture_states",
"daily_type_aggregates",
"export_manifests",
"export_items"
]
private let databaseURL: URL
private var didPrepareSchema = false
private nonisolated static let captureStateEncoder = PropertyListEncoder()
private nonisolated static let captureStateDecoder = PropertyListDecoder()
private static let monitoredTypeMetadataByIdentifier: [String: (displayName: String, category: String)] = {
Dictionary(uniqueKeysWithValues: HealthKitService.allTypes.map {
($0.id, (displayName: $0.displayName, category: $0.category.rawValue))
})
}()
init(databaseURL: URL? = nil) {
self.databaseURL = databaseURL ?? Self.defaultDatabaseURL
}
func beginObservation(observedAt: Date, triggerReason: String, selectedTypeSetHash: String?) async throws -> Int64 {
let db = try openDatabase()
defer { sqlite3_close(db) }
try prepareSchemaIfNeeded(db)
try execute("BEGIN IMMEDIATE TRANSACTION", db: db)
do {
let observationID = try createObservation(
observedAt: observedAt,
triggerReason: triggerReason,
status: "running",
selectedTypeSetHash: selectedTypeSetHash,
db: db
)
try execute("COMMIT", db: db)
return observationID
} catch {
try? execute("ROLLBACK", db: db)
throw error
}
}
func finishObservation(observationID: Int64, status: String, endedAt: Date) async throws {
let db = try openDatabase()
defer { sqlite3_close(db) }
try prepareSchemaIfNeeded(db)
try execute("BEGIN IMMEDIATE TRANSACTION", db: db)
do {
try updateObservationStatus(
observationID: observationID,
status: status,
endedAt: endedAt,
db: db
)
try execute("COMMIT", db: db)
} catch {
try? execute("ROLLBACK", db: db)
throw error
}
}
func observationRows(limit: Int = 200) async throws -> [CachedArchiveObservationRow] {
let db = try openDatabase()
defer { sqlite3_close(db) }
try prepareSchemaIfNeeded(db)
return try observationRows(limit: limit, db: db)
}
func typeSummaries(observationID: Int64, limit: Int? = nil) async throws -> [CachedArchiveTypeSummary] {
let db = try openDatabase()
defer { sqlite3_close(db) }
try prepareSchemaIfNeeded(db)
return try typeSummaries(observationID: observationID, limit: limit, db: db)
}
func typeCaptureState(sampleTypeIdentifier: String) async throws -> HealthArchiveTypeCaptureState? {
let db = try openDatabase()
defer { sqlite3_close(db) }
try prepareSchemaIfNeeded(db)
return try typeCaptureState(sampleTypeIdentifier: sampleTypeIdentifier, db: db)
}
func upsertTypeCaptureState(_ state: HealthArchiveTypeCaptureState, observationID: Int64) async throws {
let db = try openDatabase()
defer { sqlite3_close(db) }
try prepareSchemaIfNeeded(db)
try execute("BEGIN IMMEDIATE TRANSACTION", db: db)
do {
try upsertTypeCaptureState(state, observationID: observationID, db: db)
try execute("COMMIT", db: db)
} catch {
try? execute("ROLLBACK", db: db)
throw error
}
}
func upsertSamples(_ samples: [HKSample], observedAt: Date) async throws -> HealthArchiveWriteSummary {
guard !samples.isEmpty else {
return HealthArchiveWriteSummary(insertedCount: 0, updatedCount: 0, unchangedCount: 0)
}
let db = try openDatabase()
defer { sqlite3_close(db) }
try prepareSchemaIfNeeded(db)
try execute("BEGIN IMMEDIATE TRANSACTION", db: db)
do {
let summary = try upsertSamples(
samples,
observedAt: observedAt,
db: db,
rebuildDerivedState: true
)
try execute("PRAGMA foreign_key_check", db: db)
try execute("COMMIT", db: db)
return summary
} catch {
try? execute("ROLLBACK", db: db)
throw error
}
}
func upsertSamples(_ samples: [HKSample], observedAt: Date, observationID: Int64) async throws -> HealthArchiveWriteSummary {
guard !samples.isEmpty else {
return HealthArchiveWriteSummary(insertedCount: 0, updatedCount: 0, unchangedCount: 0)
}
let db = try openDatabase()
defer { sqlite3_close(db) }
try prepareSchemaIfNeeded(db)
try execute("BEGIN IMMEDIATE TRANSACTION", db: db)
do {
let summary = try upsertSamples(
samples,
observedAt: observedAt,
observationID: observationID,
db: db,
rebuildDerivedState: false
)
try execute("COMMIT", db: db)
return summary
} catch {
try? execute("ROLLBACK", db: db)
throw error
}
}
func markVerification(sampleType: HKSampleType, verifiedAt: Date) async throws {
let db = try openDatabase()
defer { sqlite3_close(db) }
try prepareSchemaIfNeeded(db)
try execute("BEGIN IMMEDIATE TRANSACTION", db: db)
do {
let observationID = try createObservation(
observedAt: verifiedAt,
triggerReason: "verification",
status: "completed",
db: db
)
try markVerification(sampleType: sampleType, verifiedAt: verifiedAt, observationID: observationID, db: db)
try execute("COMMIT", db: db)
} catch {
try? execute("ROLLBACK", db: db)
throw error
}
}
func markVerification(sampleType: HKSampleType, verifiedAt: Date, observationID: Int64) async throws {
let db = try openDatabase()
defer { sqlite3_close(db) }
try prepareSchemaIfNeeded(db)
try execute("BEGIN IMMEDIATE TRANSACTION", db: db)
do {
try markVerification(sampleType: sampleType, verifiedAt: verifiedAt, observationID: observationID, db: db)
try execute("COMMIT", db: db)
} catch {
try? execute("ROLLBACK", db: db)
throw error
}
}
func markUnchangedVerification(sampleType: HKSampleType, verifiedAt: Date, observationID: Int64) async throws {
let db = try openDatabase()
defer { sqlite3_close(db) }
try prepareSchemaIfNeeded(db)
try execute("BEGIN IMMEDIATE TRANSACTION", db: db)
do {
try markUnchangedVerification(
sampleType: sampleType,
verifiedAt: verifiedAt,
observationID: observationID,
db: db
)
try execute("COMMIT", db: db)
} catch {
try? execute("ROLLBACK", db: db)
throw error
}
}
func recordDisappearance(sampleUUIDHash: String, sampleTypeIdentifier: String, observedMissingAt: Date) async throws {
let db = try openDatabase()
defer { sqlite3_close(db) }
try prepareSchemaIfNeeded(db)
try execute("BEGIN IMMEDIATE TRANSACTION", db: db)
do {
let observationID = try createObservation(
observedAt: observedMissingAt,
triggerReason: "deleted_object",
status: "completed",
db: db
)
try recordDisappearance(
sampleUUIDHash: sampleUUIDHash,
sampleTypeIdentifier: sampleTypeIdentifier,
observedMissingAt: observedMissingAt,
observationID: observationID,
db: db,
rebuildDerivedState: true
)
try execute("COMMIT", db: db)
} catch {
try? execute("ROLLBACK", db: db)
throw error
}
}
func recordDisappearance(
sampleUUIDHash: String,
sampleTypeIdentifier: String,
observedMissingAt: Date,
observationID: Int64
) async throws {
let db = try openDatabase()
defer { sqlite3_close(db) }
try prepareSchemaIfNeeded(db)
try execute("BEGIN IMMEDIATE TRANSACTION", db: db)
do {
try recordDisappearance(
sampleUUIDHash: sampleUUIDHash,
sampleTypeIdentifier: sampleTypeIdentifier,
observedMissingAt: observedMissingAt,
observationID: observationID,
db: db,
rebuildDerivedState: false
)
try execute("COMMIT", db: db)
} catch {
try? execute("ROLLBACK", db: db)
throw error
}
}
func recordDisappearances(
sampleUUIDHashes: [String],
sampleTypeIdentifier: String,
observedMissingAt: Date,
observationID: Int64
) async throws -> Int {
guard !sampleUUIDHashes.isEmpty else { return 0 }
let db = try openDatabase()
defer { sqlite3_close(db) }
try prepareSchemaIfNeeded(db)
try execute("BEGIN IMMEDIATE TRANSACTION", db: db)
do {
let deletedCount = try recordDisappearances(
sampleUUIDHashes: sampleUUIDHashes,
sampleTypeIdentifier: sampleTypeIdentifier,
observedMissingAt: observedMissingAt,
observationID: observationID,
db: db,
rebuildDerivedState: false
)
try execute("COMMIT", db: db)
return deletedCount
} catch {
try? execute("ROLLBACK", db: db)
throw error
}
}
func records(for request: HealthArchiveRecordRequest) async throws -> [ArchivedHealthRecord] {
let db = try openDatabase()
defer { sqlite3_close(db) }
try prepareSchemaIfNeeded(db)
var clauses: [String] = []
if request.sampleTypeIdentifier != nil {
clauses.append("t.type_identifier = ?")
}
if !request.fingerprints.isEmpty {
clauses.append("s.strict_fingerprint IN (\(Array(repeating: "?", count: request.fingerprints.count).joined(separator: ",")))")
}
if request.disappearedOnly {
clauses.append("rr.last_observation_id IS NOT NULL AND es.disappeared_at IS NOT NULL")
}
if request.firstSeenAfter != nil {
clauses.append("s.first_seen_at >= ?")
}
if request.firstSeenBefore != nil {
clauses.append("s.first_seen_at <= ?")
}
if request.afterCursor != nil {
clauses.append("(v.start_date > ? OR (v.start_date = ? AND s.strict_fingerprint > ?))")
}
let whereClause = clauses.isEmpty ? "" : "WHERE \(clauses.joined(separator: " AND "))"
let limitClause = request.limit.map { "LIMIT \(max($0, 0))" } ?? ""
let sql = """
WITH selected_ranges AS (
SELECT
r.sample_id, r.version_id, r.first_observation_id, r.last_observation_id,
r.first_seen_at, r.last_seen_at,
ROW_NUMBER() OVER (
PARTITION BY r.sample_id
ORDER BY
CASE
WHEN ? IS NOT NULL THEN 0
WHEN r.last_observation_id IS NULL THEN 0
ELSE 1
END,
COALESCE(r.last_observation_id, 9223372036854775807) DESC,
r.first_observation_id DESC
) AS record_rank
FROM sample_visibility_ranges r
WHERE (? IS NULL OR (
r.first_observation_id <= ?
AND (r.last_observation_id IS NULL OR r.last_observation_id > ?)
))
),
record_ranges AS (
SELECT *
FROM selected_ranges
WHERE record_rank = 1
),
event_summary AS (
SELECT
sample_id,
MAX(CASE WHEN event_kind != 'disappeared' THEN observed_at END) AS last_seen_at,
MAX(observed_at) AS last_verified_at,
MAX(CASE WHEN event_kind = 'disappeared' THEN observed_at END) AS disappeared_at
FROM sample_observation_events
GROUP BY sample_id
)
SELECT
COALESCE(s.sample_uuid_hash, s.strict_fingerprint) AS record_id,
t.type_identifier, s.strict_fingerprint, s.semantic_fingerprint, s.sample_uuid_hash,
v.start_date, v.end_date, s.first_seen_at,
COALESCE(es.last_seen_at, rr.first_seen_at) AS last_seen_at,
es.last_verified_at,
CASE WHEN rr.last_observation_id IS NULL THEN NULL ELSE es.disappeared_at END AS disappeared_at,
v.value_kind, v.numeric_value, v.unit, v.category_value, v.workout_activity_type, v.duration_seconds,
src.bundle_identifier
FROM record_ranges rr
JOIN samples s ON s.id = rr.sample_id
JOIN sample_types t ON t.id = s.sample_type_id
JOIN sample_versions v ON v.id = rr.version_id
LEFT JOIN source_revisions sr ON sr.id = v.source_revision_id
LEFT JOIN sources src ON src.id = sr.source_id
LEFT JOIN event_summary es ON es.sample_id = s.id
\(whereClause)
ORDER BY v.start_date ASC, s.strict_fingerprint ASC
\(limitClause)
"""
return try withStatement(sql, db: db) { statement in
var index: Int32 = 1
bindInt64(request.visibleAtObservationID, to: index, in: statement)
index += 1
bindInt64(request.visibleAtObservationID, to: index, in: statement)
index += 1
bindInt64(request.visibleAtObservationID, to: index, in: statement)
index += 1
bindInt64(request.visibleAtObservationID, to: index, in: statement)
index += 1
if let typeIdentifier = request.sampleTypeIdentifier {
bindText(typeIdentifier, to: index, in: statement)
index += 1
}
for fingerprint in request.fingerprints.sorted() {
bindText(fingerprint, to: index, in: statement)
index += 1
}
if let firstSeenAfter = request.firstSeenAfter {
sqlite3_bind_double(statement, index, firstSeenAfter.timeIntervalSince1970)
index += 1
}
if let firstSeenBefore = request.firstSeenBefore {
sqlite3_bind_double(statement, index, firstSeenBefore.timeIntervalSince1970)
index += 1
}
if let cursor = request.afterCursor {
sqlite3_bind_double(statement, index, cursor.startDate.timeIntervalSince1970)
index += 1
sqlite3_bind_double(statement, index, cursor.startDate.timeIntervalSince1970)
index += 1
bindText(cursor.strictFingerprint, to: index, in: statement)
index += 1
}
var records: [ArchivedHealthRecord] = []
while sqlite3_step(statement) == SQLITE_ROW {
records.append(archiveRecord(from: statement))
}
return records
}
}
func diffSummary(_ request: HealthArchiveDiffRequest) async throws -> HealthArchiveDiffSummary {
let db = try openDatabase()
defer { sqlite3_close(db) }
try prepareSchemaIfNeeded(db)
let typeClause = request.sampleTypeIdentifier == nil ? "" : "AND t.type_identifier = ?"
let sql = """
WITH from_visible AS (
SELECT sample_id, version_id
FROM sample_visibility_ranges
WHERE first_observation_id <= ?
AND (last_observation_id IS NULL OR last_observation_id > ?)
),
to_visible AS (
SELECT sample_id, version_id
FROM sample_visibility_ranges
WHERE first_observation_id <= ?
AND (last_observation_id IS NULL OR last_observation_id > ?)
)
SELECT
(
SELECT COUNT(*)
FROM to_visible tv
LEFT JOIN from_visible fv ON fv.sample_id = tv.sample_id
JOIN samples s ON s.id = tv.sample_id
JOIN sample_types t ON t.id = s.sample_type_id
WHERE fv.sample_id IS NULL \(typeClause)
) AS appeared_count,
(
SELECT COUNT(*)
FROM from_visible fv
LEFT JOIN to_visible tv ON tv.sample_id = fv.sample_id
JOIN samples s ON s.id = fv.sample_id
JOIN sample_types t ON t.id = s.sample_type_id
WHERE tv.sample_id IS NULL \(typeClause)
) AS disappeared_count,
(
SELECT COUNT(*)
FROM to_visible tv
JOIN from_visible fv ON fv.sample_id = tv.sample_id
JOIN samples s ON s.id = tv.sample_id
JOIN sample_types t ON t.id = s.sample_type_id
WHERE tv.version_id != fv.version_id \(typeClause)
) AS representation_changed_count
"""
return try withStatement(sql, db: db) { statement in
var index: Int32 = 1
bindDiffObservationIDs(request.fromObservationID, request.toObservationID, to: statement, startingAt: &index)
for _ in 0..<3 {
if let sampleTypeIdentifier = request.sampleTypeIdentifier {
bindText(sampleTypeIdentifier, to: index, in: statement)
index += 1
}
}
guard sqlite3_step(statement) == SQLITE_ROW else {
return HealthArchiveDiffSummary(
fromObservationID: request.fromObservationID,
toObservationID: request.toObservationID,
sampleTypeIdentifier: request.sampleTypeIdentifier,
appearedCount: 0,
disappearedCount: 0,
representationChangedCount: 0
)
}
return HealthArchiveDiffSummary(
fromObservationID: request.fromObservationID,
toObservationID: request.toObservationID,
sampleTypeIdentifier: request.sampleTypeIdentifier,
appearedCount: columnInt(statement, 0) ?? 0,
disappearedCount: columnInt(statement, 1) ?? 0,
representationChangedCount: columnInt(statement, 2) ?? 0
)
}
}
func diffRecords(_ request: HealthArchiveDiffRecordRequest) async throws -> [ArchivedHealthRecord] {
let db = try openDatabase()
defer { sqlite3_close(db) }
try prepareSchemaIfNeeded(db)
let selectedRangeSQL: String
switch request.kind {
case .appeared:
selectedRangeSQL = """
SELECT tv.sample_id, tv.version_id
FROM to_visible tv
LEFT JOIN from_visible fv ON fv.sample_id = tv.sample_id
WHERE fv.sample_id IS NULL
"""
case .disappeared:
selectedRangeSQL = """
SELECT fv.sample_id, fv.version_id
FROM from_visible fv
LEFT JOIN to_visible tv ON tv.sample_id = fv.sample_id
WHERE tv.sample_id IS NULL
"""
case .representationChanged:
selectedRangeSQL = """
SELECT tv.sample_id, tv.version_id
FROM to_visible tv
JOIN from_visible fv ON fv.sample_id = tv.sample_id
WHERE tv.version_id != fv.version_id
"""
}
var clauses: [String] = []
if request.sampleTypeIdentifier != nil {
clauses.append("t.type_identifier = ?")
}
if request.afterCursor != nil {
clauses.append("(v.start_date > ? OR (v.start_date = ? AND s.strict_fingerprint > ?))")
}
let whereClause = clauses.isEmpty ? "" : "WHERE \(clauses.joined(separator: " AND "))"
let limitClause = request.limit.map { "LIMIT \(max($0, 0))" } ?? ""
let sql = """
WITH from_visible AS (
SELECT sample_id, version_id
FROM sample_visibility_ranges
WHERE first_observation_id <= ?
AND (last_observation_id IS NULL OR last_observation_id > ?)
),
to_visible AS (
SELECT sample_id, version_id
FROM sample_visibility_ranges
WHERE first_observation_id <= ?
AND (last_observation_id IS NULL OR last_observation_id > ?)
),
selected_ranges AS (
\(selectedRangeSQL)
),
event_summary AS (
SELECT
sample_id,
MAX(CASE WHEN event_kind != 'disappeared' THEN observed_at END) AS last_seen_at,
MAX(observed_at) AS last_verified_at,
MAX(CASE WHEN event_kind = 'disappeared' THEN observed_at END) AS disappeared_at
FROM sample_observation_events
WHERE observation_id <= ?
GROUP BY sample_id
)
SELECT
COALESCE(s.sample_uuid_hash, s.strict_fingerprint) AS record_id,
t.type_identifier, s.strict_fingerprint, s.semantic_fingerprint, s.sample_uuid_hash,
v.start_date, v.end_date, s.first_seen_at,
COALESCE(es.last_seen_at, s.first_seen_at) AS last_seen_at,
es.last_verified_at,
es.disappeared_at,
v.value_kind, v.numeric_value, v.unit, v.category_value, v.workout_activity_type, v.duration_seconds,
src.bundle_identifier
FROM selected_ranges srng
JOIN samples s ON s.id = srng.sample_id
JOIN sample_types t ON t.id = s.sample_type_id
JOIN sample_versions v ON v.id = srng.version_id
LEFT JOIN source_revisions sr ON sr.id = v.source_revision_id
LEFT JOIN sources src ON src.id = sr.source_id
LEFT JOIN event_summary es ON es.sample_id = s.id
\(whereClause)
ORDER BY v.start_date ASC, s.strict_fingerprint ASC
\(limitClause)
"""
return try withStatement(sql, db: db) { statement in
var index: Int32 = 1
bindDiffObservationIDs(request.fromObservationID, request.toObservationID, to: statement, startingAt: &index)
bindInt64(request.toObservationID, to: index, in: statement)
index += 1
if let sampleTypeIdentifier = request.sampleTypeIdentifier {
bindText(sampleTypeIdentifier, to: index, in: statement)
index += 1
}
if let cursor = request.afterCursor {
sqlite3_bind_double(statement, index, cursor.startDate.timeIntervalSince1970)
index += 1
sqlite3_bind_double(statement, index, cursor.startDate.timeIntervalSince1970)
index += 1
bindText(cursor.strictFingerprint, to: index, in: statement)
index += 1
}
var records: [ArchivedHealthRecord] = []
while sqlite3_step(statement) == SQLITE_ROW {
records.append(archiveRecord(from: statement))
}
return records
}
}
func aggregateComparison(_ request: HealthArchiveAggregateComparisonRequest) async throws -> [HealthArchiveAggregateComparisonRow] {
let db = try openDatabase()
defer { sqlite3_close(db) }
try prepareSchemaIfNeeded(db)
var clauses: [String] = []
if request.sampleTypeIdentifier != nil {
clauses.append("t.type_identifier = ?")
}
if request.afterBucketStart != nil {
clauses.append("k.bucket_start > ?")
}
let whereClause = clauses.isEmpty ? "" : "WHERE \(clauses.joined(separator: " AND "))"
let limitClause = request.limit.map { "LIMIT \(max($0, 0))" } ?? ""
let sql = """
WITH from_aggregates AS (
SELECT
sample_type_id,
bucket_start,
bucket_end,
SUM(visible_record_count) AS visible_record_count,
SUM(value_sum) AS value_sum
FROM daily_type_aggregates
WHERE observation_id = ?
GROUP BY sample_type_id, bucket_start, bucket_end
),
to_aggregates AS (
SELECT
sample_type_id,
bucket_start,
bucket_end,
SUM(visible_record_count) AS visible_record_count,
SUM(value_sum) AS value_sum
FROM daily_type_aggregates
WHERE observation_id = ?
GROUP BY sample_type_id, bucket_start, bucket_end
),
aggregate_keys AS (
SELECT sample_type_id, bucket_start, bucket_end FROM from_aggregates
UNION
SELECT sample_type_id, bucket_start, bucket_end FROM to_aggregates
)
SELECT
t.type_identifier,
k.bucket_start,
k.bucket_end,
COALESCE(f.visible_record_count, 0) AS from_visible_record_count,
COALESCE(ta.visible_record_count, 0) AS to_visible_record_count,
f.value_sum,
ta.value_sum
FROM aggregate_keys k
JOIN sample_types t ON t.id = k.sample_type_id
LEFT JOIN from_aggregates f
ON f.sample_type_id = k.sample_type_id
AND f.bucket_start = k.bucket_start
AND f.bucket_end = k.bucket_end
LEFT JOIN to_aggregates ta
ON ta.sample_type_id = k.sample_type_id
AND ta.bucket_start = k.bucket_start
AND ta.bucket_end = k.bucket_end
\(whereClause)
ORDER BY k.bucket_start ASC, t.type_identifier ASC
\(limitClause)
"""
return try withStatement(sql, db: db) { statement in
var index: Int32 = 1
bindInt64(request.fromObservationID, to: index, in: statement)
index += 1
bindInt64(request.toObservationID, to: index, in: statement)
index += 1
if let sampleTypeIdentifier = request.sampleTypeIdentifier {
bindText(sampleTypeIdentifier, to: index, in: statement)
index += 1
}
if let afterBucketStart = request.afterBucketStart {
sqlite3_bind_double(statement, index, afterBucketStart.timeIntervalSince1970)
index += 1
}
var rows: [HealthArchiveAggregateComparisonRow] = []
while sqlite3_step(statement) == SQLITE_ROW {
guard let bucketStart = columnUnixDate(statement, 1),
let bucketEnd = columnUnixDate(statement, 2) else {
continue
}
rows.append(HealthArchiveAggregateComparisonRow(
sampleTypeIdentifier: columnText(statement, 0) ?? "",
bucketStart: bucketStart,
bucketEnd: bucketEnd,
fromVisibleRecordCount: columnInt(statement, 3) ?? 0,
toVisibleRecordCount: columnInt(statement, 4) ?? 0,
fromValueSum: columnDouble(statement, 5),
toValueSum: columnDouble(statement, 6)
))
}
return rows
}
}
func sourceProvenanceBreakdown(_ request: HealthArchiveSourceProvenanceRequest) async throws -> [HealthArchiveSourceProvenanceRow] {
let db = try openDatabase()
defer { sqlite3_close(db) }
try prepareSchemaIfNeeded(db)
var clauses: [String] = []
if request.sampleTypeIdentifier != nil {
clauses.append("t.type_identifier = ?")
}
let whereClause = clauses.isEmpty ? "" : "WHERE \(clauses.joined(separator: " AND "))"
let limitClause = request.limit.map { "LIMIT \(max($0, 0))" } ?? ""
let sql = """
WITH visible_ranges AS (
SELECT sample_id, version_id
FROM sample_visibility_ranges
WHERE (? IS NULL OR (
first_observation_id <= ?
AND (last_observation_id IS NULL OR last_observation_id > ?)
))
)
SELECT
t.type_identifier,
src.source_name_hash,
src.bundle_identifier,
sr.product_type,
sr.version,
sr.operating_system_version,
d.model,
COUNT(*) AS visible_record_count,
SUM(v.numeric_value) AS value_sum,
MIN(v.start_date) AS earliest_start_date,
MAX(v.end_date) AS latest_end_date
FROM visible_ranges vr
JOIN sample_versions v ON v.id = vr.version_id
JOIN samples s ON s.id = vr.sample_id
JOIN sample_types t ON t.id = s.sample_type_id
LEFT JOIN source_revisions sr ON sr.id = v.source_revision_id
LEFT JOIN sources src ON src.id = sr.source_id
LEFT JOIN hk_devices d ON d.id = v.hk_device_id
\(whereClause)
GROUP BY
t.type_identifier,
src.source_name_hash,
src.bundle_identifier,
sr.product_type,
sr.version,
sr.operating_system_version,
d.model
ORDER BY visible_record_count DESC, t.type_identifier ASC, src.bundle_identifier ASC, d.model ASC
\(limitClause)
"""
return try withStatement(sql, db: db) { statement in
var index: Int32 = 1
bindInt64(request.visibleAtObservationID, to: index, in: statement)
index += 1
bindInt64(request.visibleAtObservationID, to: index, in: statement)
index += 1
bindInt64(request.visibleAtObservationID, to: index, in: statement)
index += 1
if let sampleTypeIdentifier = request.sampleTypeIdentifier {
bindText(sampleTypeIdentifier, to: index, in: statement)
index += 1
}
var rows: [HealthArchiveSourceProvenanceRow] = []
while sqlite3_step(statement) == SQLITE_ROW {
rows.append(HealthArchiveSourceProvenanceRow(
sampleTypeIdentifier: columnText(statement, 0) ?? "",
sourceNameHash: columnText(statement, 1),
sourceBundleIdentifier: columnText(statement, 2),
sourceProductType: columnText(statement, 3),
sourceVersion: columnText(statement, 4),
sourceOperatingSystemVersion: columnText(statement, 5),
deviceModel: columnText(statement, 6),
visibleRecordCount: columnInt(statement, 7) ?? 0,
valueSum: columnDouble(statement, 8),
earliestStartDate: columnUnixDate(statement, 9),
latestEndDate: columnUnixDate(statement, 10)
))
}
return rows
}
}
func consolidationEvidence(_ request: HealthArchiveConsolidationEvidenceRequest) async throws -> [HealthArchiveConsolidationEvidence] {
let db = try openDatabase()
defer { sqlite3_close(db) }
try prepareSchemaIfNeeded(db)
let typeClause = request.sampleTypeIdentifier == nil ? "" : "AND t.type_identifier = ?"
let topFilterClause = request.sampleTypeIdentifier == nil ? "" : "WHERE tk.type_identifier = ?"
let sql = """
WITH from_visible AS (
SELECT sample_id, version_id
FROM sample_visibility_ranges
WHERE first_observation_id <= ?
AND (last_observation_id IS NULL OR last_observation_id > ?)
),
to_visible AS (
SELECT sample_id, version_id
FROM sample_visibility_ranges
WHERE first_observation_id <= ?
AND (last_observation_id IS NULL OR last_observation_id > ?)
),
type_keys AS (
SELECT DISTINCT sample_type_id, type_identifier
FROM (
SELECT s.sample_type_id, t.type_identifier
FROM from_visible fv
JOIN samples s ON s.id = fv.sample_id
JOIN sample_types t ON t.id = s.sample_type_id
UNION
SELECT s.sample_type_id, t.type_identifier
FROM to_visible tv
JOIN samples s ON s.id = tv.sample_id
JOIN sample_types t ON t.id = s.sample_type_id
)
),
from_type_metrics AS (
SELECT
s.sample_type_id,
COUNT(*) AS from_visible_record_count,
SUM(v.numeric_value) AS from_value_sum,
MIN(v.start_date) AS from_start_date,
MAX(v.end_date) AS from_end_date,
COUNT(DISTINCT CASE WHEN src.bundle_identifier IS NULL THEN '__NULL__' ELSE src.bundle_identifier END) AS from_bundle_count,
COUNT(DISTINCT CASE WHEN d.model IS NULL THEN '__NULL__' ELSE d.model END) AS from_model_count
FROM from_visible fv
JOIN samples s ON s.id = fv.sample_id
JOIN sample_versions v ON v.id = fv.version_id
LEFT JOIN source_revisions sr ON sr.id = v.source_revision_id
LEFT JOIN sources src ON src.id = sr.source_id
LEFT JOIN hk_devices d ON d.id = v.hk_device_id
GROUP BY s.sample_type_id
),
to_type_metrics AS (
SELECT
s.sample_type_id,
COUNT(*) AS to_visible_record_count,
SUM(v.numeric_value) AS to_value_sum,
MIN(v.start_date) AS to_start_date,
MAX(v.end_date) AS to_end_date,
COUNT(DISTINCT CASE WHEN src.bundle_identifier IS NULL THEN '__NULL__' ELSE src.bundle_identifier END) AS to_bundle_count,
COUNT(DISTINCT CASE WHEN d.model IS NULL THEN '__NULL__' ELSE d.model END) AS to_model_count
FROM to_visible tv
JOIN samples s ON s.id = tv.sample_id
JOIN sample_versions v ON v.id = tv.version_id
LEFT JOIN source_revisions sr ON sr.id = v.source_revision_id
LEFT JOIN sources src ON src.id = sr.source_id
LEFT JOIN hk_devices d ON d.id = v.hk_device_id
GROUP BY s.sample_type_id
),
appeared_by_type AS (
SELECT s.sample_type_id, COUNT(*) AS appeared_count
FROM to_visible tv
LEFT JOIN from_visible fv ON fv.sample_id = tv.sample_id
JOIN samples s ON s.id = tv.sample_id
JOIN sample_types t ON t.id = s.sample_type_id
WHERE fv.sample_id IS NULL \(typeClause)
GROUP BY s.sample_type_id
),
disappeared_by_type AS (
SELECT s.sample_type_id, COUNT(*) AS disappeared_count
FROM from_visible fv
LEFT JOIN to_visible tv ON tv.sample_id = fv.sample_id
JOIN samples s ON s.id = fv.sample_id
JOIN sample_types t ON t.id = s.sample_type_id
WHERE tv.sample_id IS NULL \(typeClause)
GROUP BY s.sample_type_id
),
changed_by_type AS (
SELECT s.sample_type_id, COUNT(*) AS representation_changed_count
FROM to_visible tv
JOIN from_visible fv ON fv.sample_id = tv.sample_id
JOIN samples s ON s.id = tv.sample_id
JOIN sample_types t ON t.id = s.sample_type_id
WHERE tv.version_id != fv.version_id \(typeClause)
GROUP BY s.sample_type_id
)
SELECT
tk.type_identifier,
COALESCE(d.disappeared_count, 0) AS disappeared_count,
COALESCE(a.appeared_count, 0) AS appeared_count,
COALESCE(c.representation_changed_count, 0) AS representation_changed_count,
COALESCE(f.from_visible_record_count, 0) AS from_visible_record_count,
COALESCE(t.to_visible_record_count, 0) AS to_visible_record_count,
f.from_value_sum,
t.to_value_sum,
f.from_start_date,
f.from_end_date,
t.to_start_date,
t.to_end_date,
CASE
WHEN f.from_start_date IS NULL OR f.from_end_date IS NULL OR t.to_start_date IS NULL OR t.to_end_date IS NULL THEN NULL
ELSE MAX(0, MIN(f.from_end_date, t.to_end_date) - MAX(f.from_start_date, t.to_start_date))
END AS coverage_overlap_seconds,
CASE
WHEN COALESCE(f.from_visible_record_count, 0) = 0 OR f.from_start_date IS NULL OR f.from_end_date IS NULL OR f.from_end_date <= f.from_start_date THEN NULL
ELSE CAST(f.from_visible_record_count AS REAL) / ((f.from_end_date - f.from_start_date) / 3600.0)
END AS density_before,
CASE
WHEN COALESCE(t.to_visible_record_count, 0) = 0 OR t.to_start_date IS NULL OR t.to_end_date IS NULL OR t.to_end_date <= t.to_start_date THEN NULL
ELSE CAST(t.to_visible_record_count AS REAL) / ((t.to_end_date - t.to_start_date) / 3600.0)
END AS density_after,
CASE
WHEN COALESCE(f.from_bundle_count, 0) <= 1
AND COALESCE(f.from_model_count, 0) <= 1
AND COALESCE(t.to_bundle_count, 0) <= 1
AND COALESCE(t.to_model_count, 0) <= 1
THEN 1 ELSE 0
END AS source_compatible
FROM type_keys tk
LEFT JOIN from_type_metrics f ON f.sample_type_id = tk.sample_type_id
LEFT JOIN to_type_metrics t ON t.sample_type_id = tk.sample_type_id
LEFT JOIN appeared_by_type a ON a.sample_type_id = tk.sample_type_id
LEFT JOIN disappeared_by_type d ON d.sample_type_id = tk.sample_type_id
LEFT JOIN changed_by_type c ON c.sample_type_id = tk.sample_type_id
\(topFilterClause)
ORDER BY tk.type_identifier
"""
return try withStatement(sql, db: db) { statement in
var index: Int32 = 1
bindInt64(request.fromObservationID, to: index, in: statement)
index += 1
bindInt64(request.fromObservationID, to: index, in: statement)
index += 1
bindInt64(request.toObservationID, to: index, in: statement)
index += 1
bindInt64(request.toObservationID, to: index, in: statement)
index += 1
if let sampleTypeIdentifier = request.sampleTypeIdentifier {
bindText(sampleTypeIdentifier, to: index, in: statement)
index += 1
}
if let sampleTypeIdentifier = request.sampleTypeIdentifier {
bindText(sampleTypeIdentifier, to: index, in: statement)
index += 1
}
if let sampleTypeIdentifier = request.sampleTypeIdentifier {
bindText(sampleTypeIdentifier, to: index, in: statement)
index += 1
}
if request.sampleTypeIdentifier != nil {
bindText(request.sampleTypeIdentifier, to: index, in: statement)
index += 1
}
var rows: [HealthArchiveConsolidationEvidence] = []
while sqlite3_step(statement) == SQLITE_ROW {
let sampleTypeIdentifier = columnText(statement, 0) ?? ""
let disappearedCount = columnInt(statement, 1) ?? 0
let appearedCount = columnInt(statement, 2) ?? 0
let representationChangedCount = columnInt(statement, 3) ?? 0
let fromVisibleRecordCount = columnInt(statement, 4) ?? 0
let toVisibleRecordCount = columnInt(statement, 5) ?? 0
let fromValueSum = columnDouble(statement, 6)
let toValueSum = columnDouble(statement, 7)
let coverageOverlapSeconds = columnDouble(statement, 12)
let densityBefore = columnDouble(statement, 13)
let densityAfter = columnDouble(statement, 14)
let sourceCompatible = columnInt(statement, 15) == 1
let stableValue: Bool
if let fromValueSum, let toValueSum {
stableValue = abs(toValueSum - fromValueSum) <= max(1.0, abs(fromValueSum) * 0.1)
} else {
stableValue = false
}
let hasCoverageOverlap = (coverageOverlapSeconds ?? 0) > 0
let densityDoesNotIncrease: Bool
if let densityBefore, let densityAfter {
densityDoesNotIncrease = densityAfter <= densityBefore
} else {
densityDoesNotIncrease = false
}
let consolidationEvidencePresent = disappearedCount > 0
&& sourceCompatible
&& hasCoverageOverlap
&& densityDoesNotIncrease
&& stableValue
let label: String
let uncertaintyReason: String?
if representationChangedCount > 0 && disappearedCount == 0 && appearedCount == 0 {
label = "representation_changed"
uncertaintyReason = nil
} else if consolidationEvidencePresent {
label = "consolidation_likely"
uncertaintyReason = nil
} else if appearedCount > 0 && disappearedCount == 0 && representationChangedCount == 0 {
label = "appeared"
uncertaintyReason = nil
} else if appearedCount > 0 && disappearedCount > 0 && sourceCompatible && stableValue {
label = "aggregate_changed"
uncertaintyReason = nil
} else if disappearedCount == 0 && appearedCount == 0 && representationChangedCount == 0 {
label = "uncertain"
uncertaintyReason = "no structural change detected"
} else {
label = "uncertain"
uncertaintyReason = "insufficient evidence for a stable consolidation label"
}
rows.append(HealthArchiveConsolidationEvidence(
sampleTypeIdentifier: sampleTypeIdentifier,
fromObservationID: request.fromObservationID,
toObservationID: request.toObservationID,
disappearedCount: disappearedCount,
appearedCount: appearedCount,
representationChangedCount: representationChangedCount,
fromVisibleRecordCount: fromVisibleRecordCount,
toVisibleRecordCount: toVisibleRecordCount,
fromValueSum: fromValueSum,
toValueSum: toValueSum,
coverageOverlapSeconds: coverageOverlapSeconds,
densityBefore: densityBefore,
densityAfter: densityAfter,
sourceCompatible: sourceCompatible,
label: label,
uncertaintyReason: uncertaintyReason
))
}
return rows
}
}
func exportPreview(_ request: HealthArchiveReportRequest) async throws -> HealthArchiveExportPreview {
let db = try openDatabase()
defer { sqlite3_close(db) }
try prepareSchemaIfNeeded(db)
let recordCount = try await exportRecordCount(for: request, db: db)
return HealthArchiveExportPreview(
reportID: request.reportID,
title: request.title,
exportKind: "observation_records_json",
estimatedRecordCount: recordCount,
typeIdentifierFilter: request.typeIdentifierFilter,
disappearedOnly: request.disappearedOnly,
firstSeenAfter: request.firstSeenAfter,
firstSeenBefore: request.firstSeenBefore,
diffFromObservationID: request.diffFromObservationID,
diffToObservationID: request.diffToObservationID,
diffKind: request.diffKind
)
}
func exportReport(_ request: HealthArchiveReportRequest) async throws -> URL {
let exportedAt = Date()
let manifest = try await buildExportManifest(request: request, exportedAt: exportedAt)
let exportURL = URL.temporaryDirectory
.appending(path: "HealthProbe-\(request.reportID.uuidString)")
.appendingPathExtension("json")
try await writeExportReport(
request: request,
manifest: manifest,
exportedAt: exportedAt,
to: exportURL
)
try recordExportManifest(request: request, manifest: manifest)
return exportURL
}
private func recordRequest(from request: HealthArchiveReportRequest) -> HealthArchiveRecordRequest {
HealthArchiveRecordRequest(
sampleTypeIdentifier: request.typeIdentifierFilter,
fingerprints: request.includedFingerprints,
disappearedOnly: request.disappearedOnly,
firstSeenAfter: request.firstSeenAfter,
firstSeenBefore: request.firstSeenBefore,
limit: nil
)
}
private func exportRecordCount(for request: HealthArchiveReportRequest, db: OpaquePointer?) async throws -> Int {
if let fromObservationID = request.diffFromObservationID,
let toObservationID = request.diffToObservationID,
let diffKind = request.diffKind {
let summary = try await diffSummary(HealthArchiveDiffRequest(
fromObservationID: fromObservationID,
toObservationID: toObservationID,
sampleTypeIdentifier: request.typeIdentifierFilter
))
switch diffKind {
case .appeared:
return summary.appearedCount
case .disappeared:
return summary.disappearedCount
case .representationChanged:
return summary.representationChangedCount
}
}
return try exportRecordCount(for: recordRequest(from: request), db: db)
}
private func exportRecordCount(for request: HealthArchiveRecordRequest, db: OpaquePointer?) throws -> Int {
var clauses: [String] = []
if request.sampleTypeIdentifier != nil {
clauses.append("t.type_identifier = ?")
}
if !request.fingerprints.isEmpty {
clauses.append("s.strict_fingerprint IN (\(Array(repeating: "?", count: request.fingerprints.count).joined(separator: ",")))")
}
if request.disappearedOnly {
clauses.append("rr.last_observation_id IS NOT NULL AND es.disappeared_at IS NOT NULL")
}
if request.firstSeenAfter != nil {
clauses.append("s.first_seen_at >= ?")
}
if request.firstSeenBefore != nil {
clauses.append("s.first_seen_at <= ?")
}
let whereClause = clauses.isEmpty ? "" : "WHERE \(clauses.joined(separator: " AND "))"
let sql = """
WITH selected_ranges AS (
SELECT
r.sample_id, r.version_id, r.first_observation_id, r.last_observation_id,
r.first_seen_at, r.last_seen_at,
ROW_NUMBER() OVER (
PARTITION BY r.sample_id
ORDER BY
CASE
WHEN ? IS NOT NULL THEN 0
WHEN r.last_observation_id IS NULL THEN 0
ELSE 1
END,
COALESCE(r.last_observation_id, 9223372036854775807) DESC,
r.first_observation_id DESC
) AS record_rank
FROM sample_visibility_ranges r
WHERE (? IS NULL OR (
r.first_observation_id <= ?
AND (r.last_observation_id IS NULL OR r.last_observation_id > ?)
))
),
record_ranges AS (
SELECT *
FROM selected_ranges
WHERE record_rank = 1
),
event_summary AS (
SELECT
sample_id,
MAX(CASE WHEN event_kind != 'disappeared' THEN observed_at END) AS last_seen_at,
MAX(observed_at) AS last_verified_at,
MAX(CASE WHEN event_kind = 'disappeared' THEN observed_at END) AS disappeared_at
FROM sample_observation_events
GROUP BY sample_id
)
SELECT COUNT(*)
FROM record_ranges rr
JOIN samples s ON s.id = rr.sample_id
JOIN sample_types t ON t.id = s.sample_type_id
JOIN sample_versions v ON v.id = rr.version_id
LEFT JOIN event_summary es ON es.sample_id = s.id
\(whereClause)
"""
return try withStatement(sql, db: db) { statement in
var index: Int32 = 1
try bindRecordFilterValues(request, to: statement, startingAt: &index, includeCursor: false)
guard sqlite3_step(statement) == SQLITE_ROW else { return 0 }
return columnInt(statement, 0) ?? 0
}
}
private func bindRecordFilterValues(
_ request: HealthArchiveRecordRequest,
to statement: OpaquePointer?,
startingAt index: inout Int32,
includeCursor: Bool
) throws {
bindInt64(request.visibleAtObservationID, to: index, in: statement)
index += 1
bindInt64(request.visibleAtObservationID, to: index, in: statement)
index += 1
bindInt64(request.visibleAtObservationID, to: index, in: statement)
index += 1
bindInt64(request.visibleAtObservationID, to: index, in: statement)
index += 1
if let typeIdentifier = request.sampleTypeIdentifier {
bindText(typeIdentifier, to: index, in: statement)
index += 1
}
for fingerprint in request.fingerprints.sorted() {
bindText(fingerprint, to: index, in: statement)
index += 1
}
if let firstSeenAfter = request.firstSeenAfter {
sqlite3_bind_double(statement, index, firstSeenAfter.timeIntervalSince1970)
index += 1
}
if let firstSeenBefore = request.firstSeenBefore {
sqlite3_bind_double(statement, index, firstSeenBefore.timeIntervalSince1970)
index += 1
}
if includeCursor, let cursor = request.afterCursor {
sqlite3_bind_double(statement, index, cursor.startDate.timeIntervalSince1970)
index += 1
sqlite3_bind_double(statement, index, cursor.startDate.timeIntervalSince1970)
index += 1
bindText(cursor.strictFingerprint, to: index, in: statement)
index += 1
}
}
private func buildExportManifest(
request: HealthArchiveReportRequest,
exportedAt: Date
) async throws -> HealthArchiveExportManifest {
var hasher = SHA256()
hasher.update(data: Data("hp:export:manifest:v1".utf8))
hasher.update(data: Data(request.reportID.uuidString.utf8))
hasher.update(data: Data(request.title.utf8))
hasher.update(data: Data(exportedAt.timeIntervalSince1970.description.utf8))
hasher.update(data: Data((request.typeIdentifierFilter ?? "*").utf8))
hasher.update(data: Data(request.disappearedOnly.description.utf8))
if let firstSeenAfter = request.firstSeenAfter {
hasher.update(data: Data(firstSeenAfter.timeIntervalSince1970.description.utf8))
}
if let firstSeenBefore = request.firstSeenBefore {
hasher.update(data: Data(firstSeenBefore.timeIntervalSince1970.description.utf8))
}
if let diffFromObservationID = request.diffFromObservationID {
hasher.update(data: Data(diffFromObservationID.description.utf8))
}
if let diffToObservationID = request.diffToObservationID {
hasher.update(data: Data(diffToObservationID.description.utf8))
}
if let diffKind = request.diffKind {
hasher.update(data: Data(diffKind.rawValue.utf8))
}
var recordCount = 0
try await forEachReportRecord(request) { record in
recordCount += 1
let data = try JSONEncoder.healthArchive.encode(ExportRecordPayload(record))
let itemHash = Self.hexDigest(SHA256.hash(data: data))
hasher.update(data: Data(itemHash.utf8))
}
let manifestHash = Self.hexDigest(hasher.finalize())
return HealthArchiveExportManifest(
exportID: request.reportID,
exportKind: "observation_records_json",
createdAt: exportedAt,
recordCount: recordCount,
manifestHash: manifestHash
)
}
private func writeExportReport(
request: HealthArchiveReportRequest,
manifest: HealthArchiveExportManifest,
exportedAt: Date,
to exportURL: URL
) async throws {
FileManager.default.createFile(atPath: exportURL.path, contents: nil)
let handle = try FileHandle(forWritingTo: exportURL)
defer { try? handle.close() }
try handle.write(contentsOf: Data("{\n".utf8))
try writeJSONField("exportFormatVersion", value: "1", quoted: false, trailingComma: true, to: handle)
try writeJSONField("reportID", value: request.reportID.uuidString, trailingComma: true, to: handle)
try writeJSONField("title", value: request.title, trailingComma: true, to: handle)
try writeJSONField("exportedAt", value: JSONEncoder.healthArchiveDateString(exportedAt), trailingComma: true, to: handle)
try handle.write(contentsOf: Data(" \"manifest\" : {\n".utf8))
try writeJSONField("exportKind", value: manifest.exportKind, trailingComma: true, indent: " ", to: handle)
try writeJSONField("recordCount", value: "\(manifest.recordCount)", quoted: false, trailingComma: true, indent: " ", to: handle)
try writeJSONField("manifestHash", value: manifest.manifestHash, trailingComma: false, indent: " ", to: handle)
try handle.write(contentsOf: Data(" },\n".utf8))
try handle.write(contentsOf: Data(" \"records\" : [\n".utf8))
var isFirstRecord = true
try await forEachReportRecord(request) { record in
if isFirstRecord {
isFirstRecord = false
} else {
try handle.write(contentsOf: Data(",\n".utf8))
}
let data = try JSONEncoder.healthArchive.encode(ExportRecordPayload(record))
try handle.write(contentsOf: data)
}
try handle.write(contentsOf: Data("\n ]\n}\n".utf8))
}
private func forEachReportRecord(
_ request: HealthArchiveReportRequest,
_ body: (ArchivedHealthRecord) throws -> Void
) async throws {
if let fromObservationID = request.diffFromObservationID,
let toObservationID = request.diffToObservationID,
let diffKind = request.diffKind {
try await forEachDiffExportRecord(
HealthArchiveDiffRecordRequest(
fromObservationID: fromObservationID,
toObservationID: toObservationID,
sampleTypeIdentifier: request.typeIdentifierFilter,
kind: diffKind
),
body
)
return
}
try await forEachExportRecord(recordRequest(from: request), body)
}
private func writeJSONField(
_ key: String,
value: String,
quoted: Bool = true,
trailingComma: Bool,
indent: String = " ",
to handle: FileHandle
) throws {
let encodedValue: String
if quoted {
let data = try JSONEncoder.healthArchive.encode(value)
encodedValue = String(decoding: data, as: UTF8.self)
} else {
encodedValue = value
}
let comma = trailingComma ? "," : ""
try handle.write(contentsOf: Data("\(indent)\"\(key)\" : \(encodedValue)\(comma)\n".utf8))
}
private func forEachExportRecord(
_ request: HealthArchiveRecordRequest,
_ body: (ArchivedHealthRecord) throws -> Void
) async throws {
var cursor: RecordCursor?
let pageSize = 500
while true {
let page = try await records(for: HealthArchiveRecordRequest(
visibleAtObservationID: request.visibleAtObservationID,
sampleTypeIdentifier: request.sampleTypeIdentifier,
fingerprints: request.fingerprints,
disappearedOnly: request.disappearedOnly,
firstSeenAfter: request.firstSeenAfter,
firstSeenBefore: request.firstSeenBefore,
afterCursor: cursor,
limit: pageSize
))
guard !page.isEmpty else { break }
for record in page {
try body(record)
}
guard page.count == pageSize, let last = page.last else { break }
cursor = RecordCursor(startDate: last.startDate, strictFingerprint: last.strictFingerprint)
}
}
private func forEachDiffExportRecord(
_ request: HealthArchiveDiffRecordRequest,
_ body: (ArchivedHealthRecord) throws -> Void
) async throws {
var cursor: RecordCursor?
let pageSize = 500
while true {
let page = try await diffRecords(HealthArchiveDiffRecordRequest(
fromObservationID: request.fromObservationID,
toObservationID: request.toObservationID,
sampleTypeIdentifier: request.sampleTypeIdentifier,
kind: request.kind,
afterCursor: cursor,
limit: pageSize
))
guard !page.isEmpty else { break }
for record in page {
try body(record)
}
guard page.count == pageSize, let last = page.last else { break }
cursor = RecordCursor(startDate: last.startDate, strictFingerprint: last.strictFingerprint)
}
}
private func recordExportManifest(
request: HealthArchiveReportRequest,
manifest: HealthArchiveExportManifest
) throws {
let db = try openDatabase()
defer { sqlite3_close(db) }
try prepareSchemaIfNeeded(db)
let filterJSON = exportFilterJSON(request)
let sql = """
INSERT OR REPLACE INTO export_manifests (
export_id, created_at, export_kind, from_observation_id, to_observation_id,
filter_json, record_count, manifest_hash
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
"""
try withStatement(sql, db: db) { statement in
bindText(manifest.exportID.uuidString, to: 1, in: statement)
sqlite3_bind_double(statement, 2, manifest.createdAt.timeIntervalSince1970)
bindText(manifest.exportKind, to: 3, in: statement)
bindInt64(request.diffFromObservationID, to: 4, in: statement)
bindInt64(request.diffToObservationID, to: 5, in: statement)
bindText(filterJSON, to: 6, in: statement)
sqlite3_bind_int64(statement, 7, Int64(manifest.recordCount))
bindText(manifest.manifestHash, to: 8, in: statement)
guard sqlite3_step(statement) == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
}
nonisolated private func exportFilterJSON(_ request: HealthArchiveReportRequest) -> String {
let payload: [String: Any] = [
"typeIdentifier": request.typeIdentifierFilter ?? "*",
"disappearedOnly": request.disappearedOnly,
"firstSeenAfter": request.firstSeenAfter?.timeIntervalSince1970 ?? NSNull(),
"firstSeenBefore": request.firstSeenBefore?.timeIntervalSince1970 ?? NSNull(),
"diffFromObservationID": request.diffFromObservationID ?? NSNull(),
"diffToObservationID": request.diffToObservationID ?? NSNull(),
"diffKind": request.diffKind?.rawValue ?? NSNull()
]
guard
let data = try? JSONSerialization.data(withJSONObject: payload, options: [.sortedKeys]),
let json = String(data: data, encoding: .utf8)
else {
return "{}"
}
return json
}
nonisolated private static func hexDigest<D: Sequence>(_ bytes: D) -> String where D.Element == UInt8 {
bytes.map { String(format: "%02x", $0) }.joined()
}
func checkIntegrity() async throws -> HealthArchiveIntegrityReport {
let db = try openDatabase()
defer { sqlite3_close(db) }
try prepareSchemaIfNeeded(db)
let schemaVersion = try archiveSchemaVersionIfPresent(db)
let sqliteIntegrityStatus = try firstText("PRAGMA integrity_check", db: db) ?? "missing"
let foreignKeyIssueCount = try countRows("PRAGMA foreign_key_check", db: db)
let requiredTables = Set(Self.requiredArchiveV2Tables)
var missingTables = Set<String>()
for tableName in requiredTables {
if try !tableExists(tableName, db: db) {
missingTables.insert(tableName)
}
}
return HealthArchiveIntegrityReport(
schemaVersion: schemaVersion,
sqliteIntegrityStatus: sqliteIntegrityStatus,
foreignKeyIssueCount: foreignKeyIssueCount,
requiredTableNames: requiredTables,
missingTableNames: missingTables
)
}
private func openDatabase() throws -> OpaquePointer? {
try FileManager.default.createDirectory(
at: databaseURL.deletingLastPathComponent(),
withIntermediateDirectories: true
)
var db: OpaquePointer?
guard sqlite3_open_v2(databaseURL.path, &db, SQLITE_OPEN_CREATE | SQLITE_OPEN_READWRITE | SQLITE_OPEN_FULLMUTEX, nil) == SQLITE_OK else {
let message = db.map(lastErrorMessage) ?? "unable to open archive database"
sqlite3_close(db)
throw SQLiteHealthArchiveStoreError.openFailed(message)
}
sqlite3_busy_timeout(db, 5_000)
do {
try execute("PRAGMA synchronous = NORMAL", db: db)
try execute("PRAGMA temp_store = MEMORY", db: db)
try execute("PRAGMA wal_autocheckpoint = 4000", db: db)
} catch {
sqlite3_close(db)
throw error
}
return db
}
private func prepareSchemaIfNeeded(_ db: OpaquePointer?) throws {
guard !didPrepareSchema else { return }
try execute("PRAGMA journal_mode = WAL", db: db)
try execute("PRAGMA foreign_keys = ON", db: db)
let existingVersion = try archiveSchemaVersionIfPresent(db)
if let existingVersion, existingVersion > Self.archiveSchemaVersion {
throw SQLiteHealthArchiveStoreError.incompatibleSchema(existingVersion)
}
if existingVersion != Self.archiveSchemaVersion {
let needsReset = existingVersion != nil ? true : try hasUserTables(db)
if needsReset {
try resetPrototypeSchema(db)
}
}
try createArchiveV2Schema(db)
try seedArchiveMetadata(db)
didPrepareSchema = true
}
private func archiveSchemaVersionIfPresent(_ db: OpaquePointer?) throws -> Int? {
guard try tableExists("archive_metadata", db: db) else { return nil }
let sql = "SELECT value FROM archive_metadata WHERE key = 'schema_version' LIMIT 1"
return try withStatement(sql, db: db) { statement in
guard sqlite3_step(statement) == SQLITE_ROW,
let value = columnText(statement, 0) else {
return nil
}
return Int(value)
}
}
private func hasUserTables(_ db: OpaquePointer?) throws -> Bool {
let sql = """
SELECT name
FROM sqlite_master
WHERE type = 'table' AND name NOT LIKE 'sqlite_%'
LIMIT 1
"""
return try withStatement(sql, db: db) { statement in
sqlite3_step(statement) == SQLITE_ROW
}
}
private func tableExists(_ tableName: String, db: OpaquePointer?) throws -> Bool {
let sql = """
SELECT 1
FROM sqlite_master
WHERE type = 'table' AND name = ?
LIMIT 1
"""
return try withStatement(sql, db: db) { statement in
bindText(tableName, to: 1, in: statement)
return sqlite3_step(statement) == SQLITE_ROW
}
}
private func resetPrototypeSchema(_ db: OpaquePointer?) throws {
// Prototype/test installs are disposable for archive v2. Future real archives
// must use explicit migrations instead of destructive reset.
try execute("PRAGMA foreign_keys = OFF", db: db)
for objectName in try schemaObjectNames(types: ["view", "trigger"], db: db) {
try execute("DROP \(objectName.kind.uppercased()) IF EXISTS \(quotedIdentifier(objectName.name))", db: db)
}
for tableName in try schemaObjectNames(types: ["table"], db: db) {
try execute("DROP TABLE IF EXISTS \(quotedIdentifier(tableName.name))", db: db)
}
try execute("PRAGMA foreign_keys = ON", db: db)
}
private func schemaObjectNames(types: [String], db: OpaquePointer?) throws -> [(kind: String, name: String)] {
let typeList = types.map { "'\($0)'" }.joined(separator: ",")
let sql = """
SELECT type, name
FROM sqlite_master
WHERE type IN (\(typeList)) AND name NOT LIKE 'sqlite_%'
ORDER BY type, name
"""
return try withStatement(sql, db: db) { statement in
var names: [(kind: String, name: String)] = []
while sqlite3_step(statement) == SQLITE_ROW {
guard let kind = columnText(statement, 0),
let name = columnText(statement, 1) else {
continue
}
names.append((kind, name))
}
return names
}
}
private func createArchiveV2Schema(_ db: OpaquePointer?) throws {
try execute("""
CREATE TABLE IF NOT EXISTS schema_migrations (
version INTEGER PRIMARY KEY,
applied_at REAL NOT NULL,
description TEXT NOT NULL
)
""", db: db)
try execute("""
CREATE TABLE IF NOT EXISTS archive_metadata (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
)
""", db: db)
try execute("""
CREATE TABLE IF NOT EXISTS device_chains (
id INTEGER PRIMARY KEY,
device_chain_hash TEXT NOT NULL UNIQUE,
created_at REAL NOT NULL,
recovered_from_keychain INTEGER NOT NULL DEFAULT 0
)
""", db: db)
try execute("""
CREATE TABLE IF NOT EXISTS observations (
id INTEGER PRIMARY KEY,
device_chain_id INTEGER NOT NULL REFERENCES device_chains(id),
observed_at REAL NOT NULL,
started_at REAL,
ended_at REAL,
status TEXT NOT NULL,
trigger_reason TEXT NOT NULL,
app_version TEXT,
os_version TEXT,
time_zone_identifier TEXT,
time_zone_seconds_from_gmt INTEGER,
schema_version INTEGER NOT NULL,
selected_type_set_hash TEXT,
notes TEXT
)
""", db: db)
try execute("CREATE INDEX IF NOT EXISTS idx_observations_device_time ON observations(device_chain_id, observed_at)", db: db)
try execute("""
CREATE TABLE IF NOT EXISTS sample_types (
id INTEGER PRIMARY KEY,
type_identifier TEXT NOT NULL UNIQUE,
display_name TEXT,
category TEXT
)
""", db: db)
try execute("""
CREATE TABLE IF NOT EXISTS observation_type_runs (
id INTEGER PRIMARY KEY,
observation_id INTEGER NOT NULL REFERENCES observations(id),
sample_type_id INTEGER NOT NULL REFERENCES sample_types(id),
status TEXT NOT NULL,
started_at REAL,
ended_at REAL,
anchor_before BLOB,
anchor_after BLOB,
inserted_event_count INTEGER NOT NULL DEFAULT 0,
deleted_event_count INTEGER NOT NULL DEFAULT 0,
verified_visible_count INTEGER,
error_kind TEXT,
error_message_hash TEXT,
UNIQUE(observation_id, sample_type_id)
)
""", db: db)
try execute("CREATE INDEX IF NOT EXISTS idx_type_runs_type_observation ON observation_type_runs(sample_type_id, observation_id)", db: db)
try execute("""
CREATE TABLE IF NOT EXISTS sources (
id INTEGER PRIMARY KEY,
source_name_hash TEXT,
bundle_identifier TEXT
)
""", db: db)
try execute("""
CREATE TABLE IF NOT EXISTS source_revisions (
id INTEGER PRIMARY KEY,
source_id INTEGER NOT NULL REFERENCES sources(id),
product_type TEXT,
version TEXT,
operating_system_version TEXT,
UNIQUE(source_id, product_type, version, operating_system_version)
)
""", db: db)
try execute("""
CREATE TABLE IF NOT EXISTS hk_devices (
id INTEGER PRIMARY KEY,
device_hash TEXT,
manufacturer_hash TEXT,
model TEXT,
hardware_version TEXT,
firmware_version TEXT,
software_version TEXT,
local_identifier_hash TEXT,
udi_hash TEXT
)
""", db: db)
try execute("""
CREATE TABLE IF NOT EXISTS metadata_blobs (
id INTEGER PRIMARY KEY,
metadata_hash TEXT NOT NULL UNIQUE,
metadata_json TEXT NOT NULL
)
""", db: db)
try execute("""
CREATE TABLE IF NOT EXISTS samples (
id INTEGER PRIMARY KEY,
sample_type_id INTEGER NOT NULL REFERENCES sample_types(id),
sample_uuid_hash TEXT,
strict_fingerprint TEXT NOT NULL,
semantic_fingerprint TEXT,
fuzzy_key TEXT,
first_seen_observation_id INTEGER NOT NULL REFERENCES observations(id),
first_seen_at REAL NOT NULL,
UNIQUE(sample_type_id, strict_fingerprint)
)
""", db: db)
try execute("CREATE INDEX IF NOT EXISTS idx_samples_type_id ON samples(sample_type_id, id)", db: db)
try execute("CREATE INDEX IF NOT EXISTS idx_samples_type_uuid_hash ON samples(sample_type_id, sample_uuid_hash)", db: db)
try execute("DROP INDEX IF EXISTS idx_samples_uuid_hash", db: db)
try execute("DROP INDEX IF EXISTS idx_samples_type_semantic", db: db)
try execute("""
CREATE TABLE IF NOT EXISTS sample_versions (
id INTEGER PRIMARY KEY,
sample_id INTEGER NOT NULL REFERENCES samples(id),
payload_hash TEXT NOT NULL,
start_date REAL NOT NULL,
end_date REAL NOT NULL,
value_kind TEXT,
numeric_value REAL,
unit TEXT,
category_value INTEGER,
workout_activity_type INTEGER,
duration_seconds REAL,
source_revision_id INTEGER REFERENCES source_revisions(id),
hk_device_id INTEGER REFERENCES hk_devices(id),
metadata_id INTEGER REFERENCES metadata_blobs(id),
created_observation_id INTEGER NOT NULL REFERENCES observations(id),
UNIQUE(sample_id, payload_hash)
)
""", db: db)
try execute("CREATE INDEX IF NOT EXISTS idx_sample_versions_sample ON sample_versions(sample_id)", db: db)
try execute("DROP INDEX IF EXISTS idx_sample_versions_time", db: db)
try execute("""
CREATE TABLE IF NOT EXISTS sample_observation_events (
id INTEGER PRIMARY KEY,
observation_id INTEGER NOT NULL REFERENCES observations(id),
sample_id INTEGER NOT NULL REFERENCES samples(id),
version_id INTEGER REFERENCES sample_versions(id),
event_kind TEXT NOT NULL,
observed_at REAL NOT NULL,
evidence_kind TEXT,
UNIQUE(observation_id, sample_id, event_kind)
)
""", db: db)
try execute("CREATE INDEX IF NOT EXISTS idx_events_observation_kind ON sample_observation_events(observation_id, event_kind)", db: db)
try execute("CREATE INDEX IF NOT EXISTS idx_events_sample ON sample_observation_events(sample_id, observation_id)", db: db)
try execute("""
CREATE TABLE IF NOT EXISTS sample_visibility_ranges (
sample_id INTEGER NOT NULL REFERENCES samples(id),
version_id INTEGER REFERENCES sample_versions(id),
first_observation_id INTEGER NOT NULL REFERENCES observations(id),
last_observation_id INTEGER REFERENCES observations(id),
first_seen_at REAL NOT NULL,
last_seen_at REAL,
PRIMARY KEY (sample_id, version_id, first_observation_id)
)
""", db: db)
try execute("CREATE INDEX IF NOT EXISTS idx_visibility_open_ranges ON sample_visibility_ranges(last_observation_id)", db: db)
try execute("DROP INDEX IF EXISTS idx_visibility_sample_open", db: db)
try execute("CREATE INDEX IF NOT EXISTS idx_visibility_sample_open_version ON sample_visibility_ranges(sample_id, last_observation_id, version_id)", db: db)
try execute("CREATE INDEX IF NOT EXISTS idx_visibility_sample_version_open ON sample_visibility_ranges(sample_id, version_id, last_observation_id)", db: db)
try execute("CREATE INDEX IF NOT EXISTS idx_visibility_point_lookup ON sample_visibility_ranges(first_observation_id, last_observation_id)", db: db)
try execute("""
CREATE TABLE IF NOT EXISTS sample_relationships (
id INTEGER PRIMARY KEY,
observation_id INTEGER REFERENCES observations(id),
source_sample_id INTEGER NOT NULL REFERENCES samples(id),
target_sample_id INTEGER NOT NULL REFERENCES samples(id),
relationship_kind TEXT NOT NULL,
metadata_id INTEGER REFERENCES metadata_blobs(id),
UNIQUE(observation_id, source_sample_id, target_sample_id, relationship_kind)
)
""", db: db)
try execute("CREATE INDEX IF NOT EXISTS idx_relationship_source ON sample_relationships(source_sample_id, relationship_kind)", db: db)
try execute("CREATE INDEX IF NOT EXISTS idx_relationship_target ON sample_relationships(target_sample_id, relationship_kind)", db: db)
try execute("""
CREATE TABLE IF NOT EXISTS observation_type_summaries (
observation_id INTEGER NOT NULL REFERENCES observations(id),
sample_type_id INTEGER NOT NULL REFERENCES sample_types(id),
visible_record_count INTEGER NOT NULL,
appeared_count INTEGER NOT NULL DEFAULT 0,
disappeared_count INTEGER NOT NULL DEFAULT 0,
representation_changed_count INTEGER NOT NULL DEFAULT 0,
earliest_start_date REAL,
latest_end_date REAL,
value_sum REAL,
value_max REAL,
aggregate_hash TEXT,
PRIMARY KEY (observation_id, sample_type_id)
)
""", db: db)
try execute("""
CREATE TABLE IF NOT EXISTS type_capture_states (
sample_type_id INTEGER PRIMARY KEY REFERENCES sample_types(id),
observation_id INTEGER REFERENCES observations(id),
anchor_data BLOB,
record_count INTEGER NOT NULL,
content_hash TEXT NOT NULL,
earliest_start_date REAL,
latest_end_date REAL,
yearly_counts_data BLOB,
updated_at REAL NOT NULL
)
""", db: db)
try execute("CREATE INDEX IF NOT EXISTS idx_type_capture_states_observation ON type_capture_states(observation_id)", db: db)
try execute("""
CREATE TABLE IF NOT EXISTS daily_type_aggregates (
observation_id INTEGER NOT NULL REFERENCES observations(id),
sample_type_id INTEGER NOT NULL REFERENCES sample_types(id),
bucket_start REAL NOT NULL,
bucket_end REAL NOT NULL,
visible_record_count INTEGER NOT NULL,
value_sum REAL,
value_max REAL,
source_revision_id INTEGER,
aggregate_hash TEXT,
PRIMARY KEY (observation_id, sample_type_id, bucket_start, source_revision_id)
)
""", db: db)
try execute("CREATE INDEX IF NOT EXISTS idx_daily_type_bucket ON daily_type_aggregates(sample_type_id, bucket_start)", db: db)
try execute("""
CREATE TABLE IF NOT EXISTS export_manifests (
id INTEGER PRIMARY KEY,
export_id TEXT NOT NULL UNIQUE,
created_at REAL NOT NULL,
export_kind TEXT NOT NULL,
from_observation_id INTEGER REFERENCES observations(id),
to_observation_id INTEGER REFERENCES observations(id),
filter_json TEXT,
manifest_hash TEXT NOT NULL,
record_count INTEGER NOT NULL
)
""", db: db)
try execute("""
CREATE TABLE IF NOT EXISTS export_items (
export_manifest_id INTEGER NOT NULL REFERENCES export_manifests(id),
sample_id INTEGER NOT NULL REFERENCES samples(id),
version_id INTEGER REFERENCES sample_versions(id),
item_hash TEXT NOT NULL,
PRIMARY KEY (export_manifest_id, sample_id, version_id)
)
""", db: db)
}
private func seedArchiveMetadata(_ db: OpaquePointer?) throws {
try upsertMetadata(key: "schema_version", value: "\(Self.archiveSchemaVersion)", db: db)
try insertMetadataIfMissing(key: "created_at_unix", value: "\(Date().timeIntervalSince1970)", db: db)
try upsertMetadata(key: "timestamp_convention", value: "unix_seconds_utc_real", db: db)
try upsertMetadata(key: "identifier_hash_algorithm", value: "hmac-sha256-local-secret", db: db)
try upsertMetadata(key: "content_hash_algorithm", value: "sha256", db: db)
try upsertMetadata(key: "prototype_reset_policy", value: "reset_or_reinitialize_test_installs", db: db)
try withStatement(
"INSERT OR IGNORE INTO schema_migrations (version, applied_at, description) VALUES (?, ?, ?)",
db: db
) { statement in
sqlite3_bind_int64(statement, 1, sqlite3_int64(Self.archiveSchemaVersion))
sqlite3_bind_double(statement, 2, Date().timeIntervalSince1970)
bindText("Initialize archive v2 schema", to: 3, in: statement)
guard sqlite3_step(statement) == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
}
private func upsertMetadata(key: String, value: String, db: OpaquePointer?) throws {
try withStatement("INSERT OR REPLACE INTO archive_metadata (key, value) VALUES (?, ?)", db: db) { statement in
bindText(key, to: 1, in: statement)
bindText(value, to: 2, in: statement)
guard sqlite3_step(statement) == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
}
private func insertMetadataIfMissing(key: String, value: String, db: OpaquePointer?) throws {
try withStatement("INSERT OR IGNORE INTO archive_metadata (key, value) VALUES (?, ?)", db: db) { statement in
bindText(key, to: 1, in: statement)
bindText(value, to: 2, in: statement)
guard sqlite3_step(statement) == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
}
private func upsertSamples(
_ samples: [HKSample],
observedAt: Date,
db: OpaquePointer?,
rebuildDerivedState: Bool
) throws -> HealthArchiveWriteSummary {
let observationID = try createObservation(
observedAt: observedAt,
triggerReason: "anchored_page",
status: "completed",
db: db
)
return try upsertSamples(
samples,
observedAt: observedAt,
observationID: observationID,
db: db,
rebuildDerivedState: rebuildDerivedState
)
}
private func upsertSamples(
_ samples: [HKSample],
observedAt: Date,
observationID: Int64,
db: OpaquePointer?,
rebuildDerivedState: Bool
) throws -> HealthArchiveWriteSummary {
let statementCache = SQLiteStatementCache(db: db)
defer { statementCache.finalizeAll() }
let lookupCache = SQLiteWriteLookupCache()
var inserted = 0
var updated = 0
var unchanged = 0
var typeEventCounts: [Int64: (inserted: Int, deleted: Int)] = [:]
for sample in samples {
let result: ArchiveV2SampleWriteResult = try autoreleasepool {
let row = ArchiveSampleRow(sample: sample, observedAt: observedAt)
return try upsertArchiveV2Sample(
row,
observationID: observationID,
db: db,
statementCache: statementCache,
lookupCache: lookupCache
)
}
var counts = typeEventCounts[result.sampleTypeID, default: (inserted: 0, deleted: 0)]
switch result.kind {
case .inserted:
inserted += 1
counts.inserted += 1
case .updated:
updated += 1
counts.inserted += 1
case .unchanged:
unchanged += 1
}
typeEventCounts[result.sampleTypeID] = counts
}
for (sampleTypeID, eventCounts) in typeEventCounts {
try insertObservationTypeRun(
observationID: observationID,
sampleTypeID: sampleTypeID,
status: "running",
observedAt: observedAt,
insertedEventCount: eventCounts.inserted,
deletedEventCount: eventCounts.deleted,
verifiedVisibleCount: nil,
db: db,
statementCache: statementCache
)
if rebuildDerivedState {
try rebuildTypeSummary(observationID: observationID, sampleTypeID: sampleTypeID, db: db)
try rebuildDailyAggregates(
observationID: observationID,
sampleTypeID: sampleTypeID,
observedAt: observedAt,
db: db
)
}
}
return HealthArchiveWriteSummary(
insertedCount: inserted,
updatedCount: updated,
unchangedCount: unchanged
)
}
private func markVerification(
sampleType: HKSampleType,
verifiedAt: Date,
observationID: Int64,
db: OpaquePointer?
) throws {
let statementCache = SQLiteStatementCache(db: db)
defer { statementCache.finalizeAll() }
let lookupCache = SQLiteWriteLookupCache()
let sampleTypeID = try upsertSampleType(
typeIdentifier: sampleType.identifier,
db: db,
statementCache: statementCache,
lookupCache: lookupCache
)
let counts = try eventCounts(observationID: observationID, sampleTypeID: sampleTypeID, db: db)
if counts.appeared == 0,
counts.disappeared == 0,
counts.representationChanged == 0,
let previousSummary = try previousTypeSummary(
sampleTypeID: sampleTypeID,
beforeObservationID: observationID,
db: db,
statementCache: statementCache
) {
try insertObservationTypeRun(
observationID: observationID,
sampleTypeID: sampleTypeID,
status: "completed",
observedAt: verifiedAt,
insertedEventCount: 0,
deletedEventCount: 0,
verifiedVisibleCount: previousSummary.aggregate.visibleRecordCount,
db: db,
statementCache: statementCache
)
try upsertTypeSummary(
observationID: observationID,
sampleTypeID: sampleTypeID,
counts: counts,
aggregate: previousSummary.aggregate,
db: db
)
try copyDailyAggregates(
fromObservationID: previousSummary.observationID,
toObservationID: observationID,
sampleTypeID: sampleTypeID,
db: db
)
return
}
let aggregate = try visibleAggregate(sampleTypeID: sampleTypeID, db: db)
let visibleCount = aggregate.visibleRecordCount
try insertObservationTypeRun(
observationID: observationID,
sampleTypeID: sampleTypeID,
status: "completed",
observedAt: verifiedAt,
insertedEventCount: 0,
deletedEventCount: 0,
verifiedVisibleCount: visibleCount,
db: db,
statementCache: statementCache
)
try upsertTypeSummary(
observationID: observationID,
sampleTypeID: sampleTypeID,
counts: counts,
aggregate: aggregate,
db: db
)
try rebuildDailyAggregates(
observationID: observationID,
sampleTypeID: sampleTypeID,
observedAt: verifiedAt,
db: db
)
}
private func markUnchangedVerification(
sampleType: HKSampleType,
verifiedAt: Date,
observationID: Int64,
db: OpaquePointer?
) throws {
let statementCache = SQLiteStatementCache(db: db)
defer { statementCache.finalizeAll() }
let lookupCache = SQLiteWriteLookupCache()
let sampleTypeID = try upsertSampleType(
typeIdentifier: sampleType.identifier,
db: db,
statementCache: statementCache,
lookupCache: lookupCache
)
guard let previousSummary = try previousTypeSummary(
sampleTypeID: sampleTypeID,
beforeObservationID: observationID,
db: db,
statementCache: statementCache
) else {
try markVerification(
sampleType: sampleType,
verifiedAt: verifiedAt,
observationID: observationID,
db: db
)
return
}
try insertObservationTypeRun(
observationID: observationID,
sampleTypeID: sampleTypeID,
status: "completed",
observedAt: verifiedAt,
insertedEventCount: 0,
deletedEventCount: 0,
verifiedVisibleCount: previousSummary.aggregate.visibleRecordCount,
db: db,
statementCache: statementCache
)
try upsertTypeSummary(
observationID: observationID,
sampleTypeID: sampleTypeID,
counts: (appeared: 0, disappeared: 0, representationChanged: 0),
aggregate: previousSummary.aggregate,
db: db
)
try copyDailyAggregates(
fromObservationID: previousSummary.observationID,
toObservationID: observationID,
sampleTypeID: sampleTypeID,
db: db
)
}
private func recordDisappearance(
sampleUUIDHash: String,
sampleTypeIdentifier: String,
observedMissingAt: Date,
observationID: Int64,
db: OpaquePointer?,
rebuildDerivedState: Bool
) throws {
_ = try recordDisappearances(
sampleUUIDHashes: [sampleUUIDHash],
sampleTypeIdentifier: sampleTypeIdentifier,
observedMissingAt: observedMissingAt,
observationID: observationID,
db: db,
rebuildDerivedState: rebuildDerivedState
)
}
private func recordDisappearances(
sampleUUIDHashes: [String],
sampleTypeIdentifier: String,
observedMissingAt: Date,
observationID: Int64,
db: OpaquePointer?,
rebuildDerivedState: Bool
) throws -> Int {
let statementCache = SQLiteStatementCache(db: db)
defer { statementCache.finalizeAll() }
let lookupCache = SQLiteWriteLookupCache()
guard let sampleTypeID = try sampleTypeID(
typeIdentifier: sampleTypeIdentifier,
db: db,
statementCache: statementCache,
lookupCache: lookupCache
) else {
return 0
}
var deletedCount = 0
for sampleUUIDHash in sampleUUIDHashes {
guard let sampleID = try sampleID(
sampleUUIDHash: sampleUUIDHash,
sampleTypeID: sampleTypeID,
db: db,
statementCache: statementCache
) else {
continue
}
try insertObservationEvent(
observationID: observationID,
sampleID: sampleID,
versionID: nil,
eventKind: "disappeared",
evidenceKind: "deleted_object",
observedAt: observedMissingAt,
db: db,
statementCache: statementCache
)
try closeOpenVisibilityRanges(
sampleID: sampleID,
excludingVersionID: nil,
closedAtObservationID: observationID,
observedAt: observedMissingAt,
db: db,
statementCache: statementCache
)
deletedCount += 1
}
guard deletedCount > 0 else {
return 0
}
try insertObservationTypeRun(
observationID: observationID,
sampleTypeID: sampleTypeID,
status: "running",
observedAt: observedMissingAt,
insertedEventCount: 0,
deletedEventCount: deletedCount,
verifiedVisibleCount: nil,
db: db,
statementCache: statementCache
)
if rebuildDerivedState {
try rebuildTypeSummary(observationID: observationID, sampleTypeID: sampleTypeID, db: db)
try rebuildDailyAggregates(
observationID: observationID,
sampleTypeID: sampleTypeID,
observedAt: observedMissingAt,
db: db
)
}
return deletedCount
}
private func upsertArchiveV2Sample(
_ row: ArchiveSampleRow,
observationID: Int64,
db: OpaquePointer?,
statementCache: SQLiteStatementCache,
lookupCache: SQLiteWriteLookupCache
) throws -> ArchiveV2SampleWriteResult {
let sampleTypeID = try upsertSampleType(
typeIdentifier: row.typeIdentifier,
db: db,
statementCache: statementCache,
lookupCache: lookupCache
)
let sourceRevisionID = try upsertSourceRevision(
row,
db: db,
statementCache: statementCache,
lookupCache: lookupCache
)
let deviceID = try upsertDevice(
row,
db: db,
statementCache: statementCache,
lookupCache: lookupCache
)
let metadataID = try upsertMetadataBlob(
row,
db: db,
statementCache: statementCache,
lookupCache: lookupCache
)
let sampleResult = try upsertSample(
row,
sampleTypeID: sampleTypeID,
observationID: observationID,
db: db,
statementCache: statementCache
)
if sampleResult.inserted {
let versionID = try insertSampleVersion(
row,
sampleID: sampleResult.id,
sourceRevisionID: sourceRevisionID,
deviceID: deviceID,
metadataID: metadataID,
observationID: observationID,
db: db,
statementCache: statementCache
)
try insertObservationEvent(
observationID: observationID,
sampleID: sampleResult.id,
versionID: versionID,
eventKind: "appeared",
evidenceKind: "healthkit_sample",
observedAt: row.observedAt,
db: db,
statementCache: statementCache,
allowConflict: false
)
try insertOpenVisibilityRange(
sampleID: sampleResult.id,
versionID: versionID,
observationID: observationID,
observedAt: row.observedAt,
db: db,
statementCache: statementCache,
allowConflict: false
)
return ArchiveV2SampleWriteResult(sampleTypeID: sampleTypeID, kind: .inserted)
}
let versionResult = try upsertSampleVersion(
row,
sampleID: sampleResult.id,
sourceRevisionID: sourceRevisionID,
deviceID: deviceID,
metadataID: metadataID,
observationID: observationID,
db: db,
statementCache: statementCache
)
if versionResult.inserted {
try insertObservationEvent(
observationID: observationID,
sampleID: sampleResult.id,
versionID: versionResult.id,
eventKind: "representationChanged",
evidenceKind: "healthkit_sample",
observedAt: row.observedAt,
db: db,
statementCache: statementCache
)
}
if versionResult.inserted {
try closeOpenVisibilityRanges(
sampleID: sampleResult.id,
excludingVersionID: versionResult.id,
closedAtObservationID: observationID,
observedAt: row.observedAt,
db: db,
statementCache: statementCache
)
try insertOpenVisibilityRange(
sampleID: sampleResult.id,
versionID: versionResult.id,
observationID: observationID,
observedAt: row.observedAt,
db: db,
statementCache: statementCache
)
} else {
if try !hasOpenVisibilityRange(
sampleID: sampleResult.id,
versionID: versionResult.id,
db: db,
statementCache: statementCache
) {
try closeOpenVisibilityRanges(
sampleID: sampleResult.id,
excludingVersionID: versionResult.id,
closedAtObservationID: observationID,
observedAt: row.observedAt,
db: db,
statementCache: statementCache
)
try insertOpenVisibilityRange(
sampleID: sampleResult.id,
versionID: versionResult.id,
observationID: observationID,
observedAt: row.observedAt,
db: db,
statementCache: statementCache
)
}
}
return ArchiveV2SampleWriteResult(
sampleTypeID: sampleTypeID,
kind: versionResult.inserted ? .updated : .unchanged
)
}
private func createObservation(
observedAt: Date,
triggerReason: String,
status: String,
selectedTypeSetHash: String? = nil,
db: OpaquePointer?
) throws -> Int64 {
let deviceChainID = try upsertCurrentDeviceChain(db)
let timeZone = TimeZone.current
let sql = """
INSERT INTO observations (
device_chain_id, observed_at, started_at, ended_at, status, trigger_reason,
app_version, os_version, time_zone_identifier, time_zone_seconds_from_gmt,
schema_version, selected_type_set_hash, notes
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL)
"""
try withStatement(sql, db: db) { statement in
bindInt64(deviceChainID, to: 1, in: statement)
sqlite3_bind_double(statement, 2, observedAt.timeIntervalSince1970)
sqlite3_bind_double(statement, 3, observedAt.timeIntervalSince1970)
sqlite3_bind_double(statement, 4, Date().timeIntervalSince1970)
bindText(status, to: 5, in: statement)
bindText(triggerReason, to: 6, in: statement)
bindText(Bundle.main.infoDictionary?["CFBundleShortVersionString"] as? String, to: 7, in: statement)
bindText(ProcessInfo.processInfo.operatingSystemVersionString, to: 8, in: statement)
bindText(timeZone.identifier, to: 9, in: statement)
sqlite3_bind_int(statement, 10, Int32(timeZone.secondsFromGMT(for: observedAt)))
bindInt(Self.archiveSchemaVersion, to: 11, in: statement)
bindText(selectedTypeSetHash, to: 12, in: statement)
guard sqlite3_step(statement) == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
return sqlite3_last_insert_rowid(db)
}
private func observationRows(limit: Int, db: OpaquePointer?) throws -> [CachedArchiveObservationRow] {
let sql = """
SELECT
o.id,
o.observed_at,
o.status,
o.trigger_reason,
o.time_zone_identifier,
COUNT(s.sample_type_id) AS tracked_type_count,
COALESCE(SUM(s.visible_record_count), 0) AS visible_record_count,
COALESCE(SUM(s.appeared_count), 0) AS appeared_count,
COALESCE(SUM(s.disappeared_count), 0) AS disappeared_count,
COALESCE(SUM(s.representation_changed_count), 0) AS representation_changed_count,
o.schema_version
FROM observations o
LEFT JOIN observation_type_summaries s ON s.observation_id = o.id
GROUP BY
o.id, o.observed_at, o.status, o.trigger_reason, o.time_zone_identifier,
o.schema_version
ORDER BY o.id DESC
LIMIT ?
"""
var rows: [CachedArchiveObservationRow] = []
try withStatement(sql, db: db) { statement in
bindInt(max(limit, 0), to: 1, in: statement)
var stepResult = sqlite3_step(statement)
while stepResult == SQLITE_ROW {
rows.append(CachedArchiveObservationRow(
observationID: sqlite3_column_int64(statement, 0),
observedAt: columnUnixDate(statement, 1) ?? Date(timeIntervalSince1970: 0),
status: columnText(statement, 2) ?? "",
triggerReason: columnText(statement, 3) ?? "",
timeZoneIdentifier: columnText(statement, 4),
trackedTypeCount: columnInt(statement, 5) ?? 0,
visibleRecordCount: columnInt(statement, 6) ?? 0,
appearedCount: columnInt(statement, 7) ?? 0,
disappearedCount: columnInt(statement, 8) ?? 0,
representationChangedCount: columnInt(statement, 9) ?? 0,
archiveSchemaVersion: columnInt(statement, 10) ?? Self.archiveSchemaVersion,
cacheSchemaVersion: CoreDataArchiveCacheStore.cacheSchemaVersion,
computedAt: Date()
))
stepResult = sqlite3_step(statement)
}
guard stepResult == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
return rows
}
private func typeSummaries(
observationID: Int64,
limit: Int?,
db: OpaquePointer?
) throws -> [CachedArchiveTypeSummary] {
let limitClause = limit == nil ? "" : "LIMIT ?"
let sql = """
SELECT
s.observation_id,
t.type_identifier,
t.display_name,
s.visible_record_count,
s.appeared_count,
s.disappeared_count,
s.representation_changed_count,
s.earliest_start_date,
s.latest_end_date,
s.value_sum,
s.value_max,
s.aggregate_hash
FROM observation_type_summaries s
JOIN sample_types t ON t.id = s.sample_type_id
WHERE s.observation_id = ?
ORDER BY t.display_name, t.type_identifier
\(limitClause)
"""
var rows: [CachedArchiveTypeSummary] = []
try withStatement(sql, db: db) { statement in
bindInt64(observationID, to: 1, in: statement)
if let limit {
bindInt(max(limit, 0), to: 2, in: statement)
}
var stepResult = sqlite3_step(statement)
while stepResult == SQLITE_ROW {
let typeIdentifier = columnText(statement, 1) ?? ""
rows.append(CachedArchiveTypeSummary(
observationID: sqlite3_column_int64(statement, 0),
sampleTypeIdentifier: typeIdentifier,
displayName: columnText(statement, 2) ?? Self.monitoredTypeMetadataByIdentifier[typeIdentifier]?.displayName,
visibleRecordCount: columnInt(statement, 3) ?? 0,
appearedCount: columnInt(statement, 4) ?? 0,
disappearedCount: columnInt(statement, 5) ?? 0,
representationChangedCount: columnInt(statement, 6) ?? 0,
earliestStartDate: columnUnixDate(statement, 7),
latestEndDate: columnUnixDate(statement, 8),
valueSum: columnDouble(statement, 9),
valueMax: columnDouble(statement, 10),
aggregateHash: columnText(statement, 11),
computedAt: Date()
))
stepResult = sqlite3_step(statement)
}
guard stepResult == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
return rows
}
private func typeCaptureState(
sampleTypeIdentifier: String,
db: OpaquePointer?
) throws -> HealthArchiveTypeCaptureState? {
let sql = """
SELECT
t.type_identifier,
s.observation_id,
s.record_count,
s.content_hash,
s.earliest_start_date,
s.latest_end_date,
s.yearly_counts_data,
s.anchor_data
FROM type_capture_states s
JOIN sample_types t ON t.id = s.sample_type_id
WHERE t.type_identifier = ?
LIMIT 1
"""
return try withStatement(sql, db: db) { statement in
bindText(sampleTypeIdentifier, to: 1, in: statement)
guard sqlite3_step(statement) == SQLITE_ROW else { return nil }
let yearlyCountsData = columnData(statement, 6)
let yearlyCounts = yearlyCountsData.flatMap {
try? Self.captureStateDecoder.decode([Int: Int].self, from: $0)
} ?? [:]
return HealthArchiveTypeCaptureState(
sampleTypeIdentifier: columnText(statement, 0) ?? sampleTypeIdentifier,
observationID: columnInt64(statement, 1),
count: columnInt(statement, 2) ?? 0,
contentHash: columnText(statement, 3) ?? "",
earliestDate: columnUnixDate(statement, 4),
latestDate: columnUnixDate(statement, 5),
yearlyCounts: yearlyCounts,
anchorData: columnData(statement, 7)
)
}
}
private func upsertTypeCaptureState(
_ state: HealthArchiveTypeCaptureState,
observationID: Int64,
db: OpaquePointer?
) throws {
let sampleTypeID = try upsertSampleType(typeIdentifier: state.sampleTypeIdentifier, db: db)
let yearlyCountsData = try? Self.captureStateEncoder.encode(state.yearlyCounts)
let sql = """
INSERT INTO type_capture_states (
sample_type_id,
observation_id,
anchor_data,
record_count,
content_hash,
earliest_start_date,
latest_end_date,
yearly_counts_data,
updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(sample_type_id) DO UPDATE SET
observation_id = excluded.observation_id,
anchor_data = excluded.anchor_data,
record_count = excluded.record_count,
content_hash = excluded.content_hash,
earliest_start_date = excluded.earliest_start_date,
latest_end_date = excluded.latest_end_date,
yearly_counts_data = excluded.yearly_counts_data,
updated_at = excluded.updated_at
"""
try withStatement(sql, db: db) { statement in
bindInt64(sampleTypeID, to: 1, in: statement)
bindInt64(observationID, to: 2, in: statement)
bindData(state.anchorData, to: 3, in: statement)
bindInt(state.count, to: 4, in: statement)
bindText(state.contentHash, to: 5, in: statement)
bindDouble(state.earliestDate?.timeIntervalSince1970, to: 6, in: statement)
bindDouble(state.latestDate?.timeIntervalSince1970, to: 7, in: statement)
bindData(yearlyCountsData, to: 8, in: statement)
sqlite3_bind_double(statement, 9, Date().timeIntervalSince1970)
guard sqlite3_step(statement) == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
}
private func updateObservationStatus(
observationID: Int64,
status: String,
endedAt: Date,
db: OpaquePointer?
) throws {
try withStatement(
"UPDATE observations SET status = ?, ended_at = ? WHERE id = ?",
db: db
) { statement in
bindText(status, to: 1, in: statement)
sqlite3_bind_double(statement, 2, endedAt.timeIntervalSince1970)
bindInt64(observationID, to: 3, in: statement)
guard sqlite3_step(statement) == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
}
private func insertObservationTypeRun(
observationID: Int64,
sampleTypeID: Int64,
status: String,
observedAt: Date,
insertedEventCount: Int,
deletedEventCount: Int,
verifiedVisibleCount: Int?,
db: OpaquePointer?,
statementCache: SQLiteStatementCache? = nil
) throws {
if let existing = try observationTypeRunCounts(
observationID: observationID,
sampleTypeID: sampleTypeID,
db: db,
statementCache: statementCache
) {
try withStatement(
"""
UPDATE observation_type_runs
SET status = ?,
started_at = COALESCE(started_at, ?),
ended_at = ?,
inserted_event_count = ?,
deleted_event_count = ?,
verified_visible_count = COALESCE(?, verified_visible_count)
WHERE observation_id = ? AND sample_type_id = ?
""",
db: db
, statementCache: statementCache) { statement in
bindText(status, to: 1, in: statement)
sqlite3_bind_double(statement, 2, observedAt.timeIntervalSince1970)
sqlite3_bind_double(statement, 3, observedAt.timeIntervalSince1970)
bindInt(existing.inserted + insertedEventCount, to: 4, in: statement)
bindInt(existing.deleted + deletedEventCount, to: 5, in: statement)
bindInt(verifiedVisibleCount, to: 6, in: statement)
bindInt64(observationID, to: 7, in: statement)
bindInt64(sampleTypeID, to: 8, in: statement)
guard sqlite3_step(statement) == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
} else {
let sql = """
INSERT INTO observation_type_runs (
observation_id, sample_type_id, status, started_at, ended_at,
inserted_event_count, deleted_event_count, verified_visible_count
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
"""
try withStatement(sql, db: db, statementCache: statementCache) { statement in
bindInt64(observationID, to: 1, in: statement)
bindInt64(sampleTypeID, to: 2, in: statement)
bindText(status, to: 3, in: statement)
sqlite3_bind_double(statement, 4, observedAt.timeIntervalSince1970)
sqlite3_bind_double(statement, 5, observedAt.timeIntervalSince1970)
bindInt(insertedEventCount, to: 6, in: statement)
bindInt(deletedEventCount, to: 7, in: statement)
bindInt(verifiedVisibleCount, to: 8, in: statement)
guard sqlite3_step(statement) == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
}
}
private func observationTypeRunCounts(
observationID: Int64,
sampleTypeID: Int64,
db: OpaquePointer?,
statementCache: SQLiteStatementCache? = nil
) throws -> (inserted: Int, deleted: Int)? {
try withStatement(
"""
SELECT inserted_event_count, deleted_event_count
FROM observation_type_runs
WHERE observation_id = ? AND sample_type_id = ?
LIMIT 1
""",
db: db,
statementCache: statementCache
) { statement in
bindInt64(observationID, to: 1, in: statement)
bindInt64(sampleTypeID, to: 2, in: statement)
guard sqlite3_step(statement) == SQLITE_ROW else { return nil }
return (Int(sqlite3_column_int(statement, 0)), Int(sqlite3_column_int(statement, 1)))
}
}
private func upsertCurrentDeviceChain(_ db: OpaquePointer?) throws -> Int64 {
let resolution = KeychainService.resolveDeviceID(swiftDataStoreIsEmpty: false)
let chainHash = HashService.archiveContentHash(domain: "hp:v2:device_chain", parts: [resolution.id])
try withStatement(
"INSERT OR IGNORE INTO device_chains (device_chain_hash, created_at, recovered_from_keychain) VALUES (?, ?, ?)",
db: db
) { statement in
bindText(chainHash, to: 1, in: statement)
sqlite3_bind_double(statement, 2, Date().timeIntervalSince1970)
sqlite3_bind_int(statement, 3, resolution.isRecovered ? 1 : 0)
guard sqlite3_step(statement) == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
return try requiredInt64(
"SELECT id FROM device_chains WHERE device_chain_hash = ? LIMIT 1",
db: db
) { statement in
bindText(chainHash, to: 1, in: statement)
}
}
private func upsertSampleType(
typeIdentifier: String,
db: OpaquePointer?,
statementCache: SQLiteStatementCache? = nil,
lookupCache: SQLiteWriteLookupCache? = nil
) throws -> Int64 {
if let cached = lookupCache?.sampleTypeIDs[typeIdentifier] {
return cached
}
let metadata = Self.monitoredTypeMetadataByIdentifier[typeIdentifier]
try withStatement(
"""
INSERT INTO sample_types (type_identifier, display_name, category)
VALUES (?, ?, ?)
ON CONFLICT(type_identifier) DO UPDATE SET
display_name = COALESCE(sample_types.display_name, excluded.display_name),
category = COALESCE(sample_types.category, excluded.category)
""",
db: db,
statementCache: statementCache
) { statement in
bindText(typeIdentifier, to: 1, in: statement)
bindText(metadata?.displayName, to: 2, in: statement)
bindText(metadata?.category, to: 3, in: statement)
guard sqlite3_step(statement) == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
let id = try requiredInt64(
"SELECT id FROM sample_types WHERE type_identifier = ? LIMIT 1",
db: db,
statementCache: statementCache
) { statement in
bindText(typeIdentifier, to: 1, in: statement)
}
lookupCache?.sampleTypeIDs[typeIdentifier] = id
return id
}
private func sampleTypeID(
typeIdentifier: String,
db: OpaquePointer?,
statementCache: SQLiteStatementCache? = nil,
lookupCache: SQLiteWriteLookupCache? = nil
) throws -> Int64? {
if let cached = lookupCache?.sampleTypeIDs[typeIdentifier] {
return cached
}
let id = try optionalInt64(
"SELECT id FROM sample_types WHERE type_identifier = ? LIMIT 1",
db: db,
statementCache: statementCache
) { statement in
bindText(typeIdentifier, to: 1, in: statement)
}
if let id {
lookupCache?.sampleTypeIDs[typeIdentifier] = id
}
return id
}
private func upsertSourceRevision(
_ row: ArchiveSampleRow,
db: OpaquePointer?,
statementCache: SQLiteStatementCache? = nil,
lookupCache: SQLiteWriteLookupCache? = nil
) throws -> Int64? {
guard row.sourceName != nil || row.sourceBundleIdentifier != nil else { return nil }
let sourceNameHash = row.sourceName.map { HashService.archiveContentHash(domain: "hp:v2:source_name", parts: [$0]) }
let key = SQLiteWriteLookupSourceRevisionKey(
sourceNameHash: sourceNameHash,
bundleIdentifier: row.sourceBundleIdentifier,
productType: row.sourceProductType,
version: row.sourceVersion,
operatingSystemVersion: row.sourceOperatingSystemVersion
)
if let cached = lookupCache?.sourceRevisionIDs[key] {
return cached
}
let sourceID = try upsertSource(
sourceNameHash: sourceNameHash,
bundleIdentifier: row.sourceBundleIdentifier,
db: db,
statementCache: statementCache,
lookupCache: lookupCache
)
if let existing = try sourceRevisionID(
sourceID: sourceID,
productType: row.sourceProductType,
version: row.sourceVersion,
operatingSystemVersion: row.sourceOperatingSystemVersion,
db: db,
statementCache: statementCache
) {
lookupCache?.sourceRevisionIDs[key] = existing
return existing
}
try withStatement(
"INSERT INTO source_revisions (source_id, product_type, version, operating_system_version) VALUES (?, ?, ?, ?)",
db: db,
statementCache: statementCache
) { statement in
bindInt64(sourceID, to: 1, in: statement)
bindText(row.sourceProductType, to: 2, in: statement)
bindText(row.sourceVersion, to: 3, in: statement)
bindText(row.sourceOperatingSystemVersion, to: 4, in: statement)
guard sqlite3_step(statement) == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
let id = sqlite3_last_insert_rowid(db)
lookupCache?.sourceRevisionIDs[key] = id
return id
}
private func sourceRevisionID(
sourceID: Int64,
productType: String?,
version: String?,
operatingSystemVersion: String?,
db: OpaquePointer?,
statementCache: SQLiteStatementCache? = nil
) throws -> Int64? {
try optionalInt64(
"""
SELECT id FROM source_revisions
WHERE source_id = ?
AND (product_type = ? OR (product_type IS NULL AND ? IS NULL))
AND (version = ? OR (version IS NULL AND ? IS NULL))
AND (operating_system_version = ? OR (operating_system_version IS NULL AND ? IS NULL))
LIMIT 1
""",
db: db,
statementCache: statementCache
) { statement in
bindInt64(sourceID, to: 1, in: statement)
bindText(productType, to: 2, in: statement)
bindText(productType, to: 3, in: statement)
bindText(version, to: 4, in: statement)
bindText(version, to: 5, in: statement)
bindText(operatingSystemVersion, to: 6, in: statement)
bindText(operatingSystemVersion, to: 7, in: statement)
}
}
private func upsertSource(
sourceNameHash: String?,
bundleIdentifier: String?,
db: OpaquePointer?,
statementCache: SQLiteStatementCache? = nil,
lookupCache: SQLiteWriteLookupCache? = nil
) throws -> Int64 {
let key = SQLiteWriteLookupSourceKey(
sourceNameHash: sourceNameHash,
bundleIdentifier: bundleIdentifier
)
if let cached = lookupCache?.sourceIDs[key] {
return cached
}
if let existing = try optionalInt64(
"""
SELECT id FROM sources
WHERE (source_name_hash = ? OR (source_name_hash IS NULL AND ? IS NULL))
AND (bundle_identifier = ? OR (bundle_identifier IS NULL AND ? IS NULL))
LIMIT 1
""",
db: db,
statementCache: statementCache,
bind: { statement in
bindText(sourceNameHash, to: 1, in: statement)
bindText(sourceNameHash, to: 2, in: statement)
bindText(bundleIdentifier, to: 3, in: statement)
bindText(bundleIdentifier, to: 4, in: statement)
}
) {
lookupCache?.sourceIDs[key] = existing
return existing
}
try withStatement(
"INSERT INTO sources (source_name_hash, bundle_identifier) VALUES (?, ?)",
db: db,
statementCache: statementCache
) { statement in
bindText(sourceNameHash, to: 1, in: statement)
bindText(bundleIdentifier, to: 2, in: statement)
guard sqlite3_step(statement) == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
let id = sqlite3_last_insert_rowid(db)
lookupCache?.sourceIDs[key] = id
return id
}
private func upsertDevice(
_ row: ArchiveSampleRow,
db: OpaquePointer?,
statementCache: SQLiteStatementCache? = nil,
lookupCache: SQLiteWriteLookupCache? = nil
) throws -> Int64? {
guard row.hasDeviceProvenance else { return nil }
let deviceHash = row.deviceName.map { HashService.archiveContentHash(domain: "hp:v2:device_name", parts: [$0]) }
let manufacturerHash = row.deviceManufacturer.map { HashService.archiveContentHash(domain: "hp:v2:device_manufacturer", parts: [$0]) }
let localIdentifierHash = row.deviceLocalIdentifier.map { HashService.archiveContentHash(domain: "hp:v2:device_local_id", parts: [$0]) }
let udiHash = row.deviceUDI.map { HashService.archiveContentHash(domain: "hp:v2:device_udi", parts: [$0]) }
let key = SQLiteWriteLookupDeviceKey(
deviceHash: deviceHash,
localIdentifierHash: localIdentifierHash,
udiHash: udiHash,
model: row.deviceModel
)
if let cached = lookupCache?.deviceIDs[key] {
return cached
}
if let existing = try optionalInt64(
"""
SELECT id FROM hk_devices
WHERE (device_hash = ? OR (device_hash IS NULL AND ? IS NULL))
AND (local_identifier_hash = ? OR (local_identifier_hash IS NULL AND ? IS NULL))
AND (udi_hash = ? OR (udi_hash IS NULL AND ? IS NULL))
AND (model = ? OR (model IS NULL AND ? IS NULL))
LIMIT 1
""",
db: db,
statementCache: statementCache,
bind: { statement in
bindText(deviceHash, to: 1, in: statement)
bindText(deviceHash, to: 2, in: statement)
bindText(localIdentifierHash, to: 3, in: statement)
bindText(localIdentifierHash, to: 4, in: statement)
bindText(udiHash, to: 5, in: statement)
bindText(udiHash, to: 6, in: statement)
bindText(row.deviceModel, to: 7, in: statement)
bindText(row.deviceModel, to: 8, in: statement)
}
) {
lookupCache?.deviceIDs[key] = existing
return existing
}
try withStatement(
"""
INSERT INTO hk_devices (
device_hash, manufacturer_hash, model, hardware_version, firmware_version,
software_version, local_identifier_hash, udi_hash
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
db: db,
statementCache: statementCache
) { statement in
bindText(deviceHash, to: 1, in: statement)
bindText(manufacturerHash, to: 2, in: statement)
bindText(row.deviceModel, to: 3, in: statement)
bindText(row.deviceHardwareVersion, to: 4, in: statement)
bindText(row.deviceFirmwareVersion, to: 5, in: statement)
bindText(row.deviceSoftwareVersion, to: 6, in: statement)
bindText(localIdentifierHash, to: 7, in: statement)
bindText(udiHash, to: 8, in: statement)
guard sqlite3_step(statement) == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
let id = sqlite3_last_insert_rowid(db)
lookupCache?.deviceIDs[key] = id
return id
}
private func upsertMetadataBlob(
_ row: ArchiveSampleRow,
db: OpaquePointer?,
statementCache: SQLiteStatementCache? = nil,
lookupCache: SQLiteWriteLookupCache? = nil
) throws -> Int64? {
guard let metadataHash = row.metadataHash, let metadataJSON = row.metadataJSON else { return nil }
if let cached = lookupCache?.metadataBlobIDs[metadataHash] {
return cached
}
try withStatement(
"INSERT OR IGNORE INTO metadata_blobs (metadata_hash, metadata_json) VALUES (?, ?)",
db: db,
statementCache: statementCache
) { statement in
bindText(metadataHash, to: 1, in: statement)
bindText(metadataJSON, to: 2, in: statement)
guard sqlite3_step(statement) == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
let id = try requiredInt64(
"SELECT id FROM metadata_blobs WHERE metadata_hash = ? LIMIT 1",
db: db,
statementCache: statementCache
) { statement in
bindText(metadataHash, to: 1, in: statement)
}
lookupCache?.metadataBlobIDs[metadataHash] = id
return id
}
private func upsertSample(
_ row: ArchiveSampleRow,
sampleTypeID: Int64,
observationID: Int64,
db: OpaquePointer?,
statementCache: SQLiteStatementCache? = nil
) throws -> (id: Int64, inserted: Bool) {
try withStatement(
"""
INSERT OR IGNORE INTO samples (
sample_type_id, sample_uuid_hash, strict_fingerprint, semantic_fingerprint,
fuzzy_key, first_seen_observation_id, first_seen_at
) VALUES (?, ?, ?, ?, NULL, ?, ?)
""",
db: db,
statementCache: statementCache
) { statement in
bindInt64(sampleTypeID, to: 1, in: statement)
bindText(row.sampleUUIDHash, to: 2, in: statement)
bindText(row.strictFingerprint, to: 3, in: statement)
bindText(row.semanticFingerprint, to: 4, in: statement)
bindInt64(observationID, to: 5, in: statement)
sqlite3_bind_double(statement, 6, row.observedAt.timeIntervalSince1970)
guard sqlite3_step(statement) == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
let inserted = sqlite3_changes(db) > 0
let id: Int64
if inserted {
id = sqlite3_last_insert_rowid(db)
} else {
id = try requiredInt64(
"SELECT id FROM samples WHERE sample_type_id = ? AND strict_fingerprint = ? LIMIT 1",
db: db,
statementCache: statementCache
) { statement in
bindInt64(sampleTypeID, to: 1, in: statement)
bindText(row.strictFingerprint, to: 2, in: statement)
}
}
return (id, inserted)
}
private func sampleID(sampleUUIDHash: String, sampleTypeID: Int64, db: OpaquePointer?, statementCache: SQLiteStatementCache? = nil) throws -> Int64? {
try optionalInt64(
"SELECT id FROM samples WHERE sample_type_id = ? AND sample_uuid_hash = ? LIMIT 1",
db: db,
statementCache: statementCache
) { statement in
bindInt64(sampleTypeID, to: 1, in: statement)
bindText(sampleUUIDHash, to: 2, in: statement)
}
}
private func upsertSampleVersion(
_ row: ArchiveSampleRow,
sampleID: Int64,
sourceRevisionID: Int64?,
deviceID: Int64?,
metadataID: Int64?,
observationID: Int64,
db: OpaquePointer?,
statementCache: SQLiteStatementCache? = nil
) throws -> (id: Int64, inserted: Bool) {
try withStatement(
"""
INSERT OR IGNORE INTO sample_versions (
sample_id, payload_hash, start_date, end_date, value_kind, numeric_value,
unit, category_value, workout_activity_type, duration_seconds,
source_revision_id, hk_device_id, metadata_id, created_observation_id
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
db: db,
statementCache: statementCache
) { statement in
bindInt64(sampleID, to: 1, in: statement)
bindText(row.payloadHash, to: 2, in: statement)
sqlite3_bind_double(statement, 3, row.startDate.timeIntervalSince1970)
sqlite3_bind_double(statement, 4, row.endDate.timeIntervalSince1970)
bindText(row.valueKind, to: 5, in: statement)
bindDouble(row.value, to: 6, in: statement)
bindText(row.unit, to: 7, in: statement)
bindInt(row.categoryValue, to: 8, in: statement)
bindInt(row.workoutActivityType, to: 9, in: statement)
bindDouble(row.durationSeconds, to: 10, in: statement)
bindInt64(sourceRevisionID, to: 11, in: statement)
bindInt64(deviceID, to: 12, in: statement)
bindInt64(metadataID, to: 13, in: statement)
bindInt64(observationID, to: 14, in: statement)
guard sqlite3_step(statement) == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
let inserted = sqlite3_changes(db) > 0
let id: Int64
if inserted {
id = sqlite3_last_insert_rowid(db)
} else {
id = try requiredInt64(
"SELECT id FROM sample_versions WHERE sample_id = ? AND payload_hash = ? LIMIT 1",
db: db,
statementCache: statementCache
) { statement in
bindInt64(sampleID, to: 1, in: statement)
bindText(row.payloadHash, to: 2, in: statement)
}
}
return (id, inserted)
}
private func insertSampleVersion(
_ row: ArchiveSampleRow,
sampleID: Int64,
sourceRevisionID: Int64?,
deviceID: Int64?,
metadataID: Int64?,
observationID: Int64,
db: OpaquePointer?,
statementCache: SQLiteStatementCache? = nil
) throws -> Int64 {
try withStatement(
"""
INSERT INTO sample_versions (
sample_id, payload_hash, start_date, end_date, value_kind, numeric_value,
unit, category_value, workout_activity_type, duration_seconds,
source_revision_id, hk_device_id, metadata_id, created_observation_id
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
db: db,
statementCache: statementCache
) { statement in
bindInt64(sampleID, to: 1, in: statement)
bindText(row.payloadHash, to: 2, in: statement)
sqlite3_bind_double(statement, 3, row.startDate.timeIntervalSince1970)
sqlite3_bind_double(statement, 4, row.endDate.timeIntervalSince1970)
bindText(row.valueKind, to: 5, in: statement)
bindDouble(row.value, to: 6, in: statement)
bindText(row.unit, to: 7, in: statement)
bindInt(row.categoryValue, to: 8, in: statement)
bindInt(row.workoutActivityType, to: 9, in: statement)
bindDouble(row.durationSeconds, to: 10, in: statement)
bindInt64(sourceRevisionID, to: 11, in: statement)
bindInt64(deviceID, to: 12, in: statement)
bindInt64(metadataID, to: 13, in: statement)
bindInt64(observationID, to: 14, in: statement)
guard sqlite3_step(statement) == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
return sqlite3_last_insert_rowid(db)
}
private func insertObservationEvent(
observationID: Int64,
sampleID: Int64,
versionID: Int64?,
eventKind: String,
evidenceKind: String,
observedAt: Date,
db: OpaquePointer?,
statementCache: SQLiteStatementCache? = nil,
allowConflict: Bool = true
) throws {
let insertMode = allowConflict ? "INSERT OR IGNORE" : "INSERT"
try withStatement(
"""
\(insertMode) INTO sample_observation_events (
observation_id, sample_id, version_id, event_kind, observed_at, evidence_kind
) VALUES (?, ?, ?, ?, ?, ?)
""",
db: db,
statementCache: statementCache
) { statement in
bindInt64(observationID, to: 1, in: statement)
bindInt64(sampleID, to: 2, in: statement)
bindInt64(versionID, to: 3, in: statement)
bindText(eventKind, to: 4, in: statement)
sqlite3_bind_double(statement, 5, observedAt.timeIntervalSince1970)
bindText(evidenceKind, to: 6, in: statement)
guard sqlite3_step(statement) == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
}
private func closeOpenVisibilityRanges(
sampleID: Int64,
excludingVersionID: Int64?,
closedAtObservationID: Int64,
observedAt: Date,
db: OpaquePointer?,
statementCache: SQLiteStatementCache? = nil
) throws {
let versionPredicate = excludingVersionID == nil
? ""
: "AND (version_id IS NULL OR version_id != ?)"
let sql = """
UPDATE sample_visibility_ranges
SET last_observation_id = ?, last_seen_at = ?
WHERE sample_id = ? AND last_observation_id IS NULL \(versionPredicate)
"""
try withStatement(sql, db: db, statementCache: statementCache) { statement in
bindInt64(closedAtObservationID, to: 1, in: statement)
sqlite3_bind_double(statement, 2, observedAt.timeIntervalSince1970)
bindInt64(sampleID, to: 3, in: statement)
if let excludingVersionID {
bindInt64(excludingVersionID, to: 4, in: statement)
}
guard sqlite3_step(statement) == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
}
private func hasOpenVisibilityRange(
sampleID: Int64,
versionID: Int64,
db: OpaquePointer?,
statementCache: SQLiteStatementCache? = nil
) throws -> Bool {
let existing = try optionalInt64(
"""
SELECT 1
FROM sample_visibility_ranges
WHERE sample_id = ? AND version_id = ? AND last_observation_id IS NULL
LIMIT 1
""",
db: db,
statementCache: statementCache
) { statement in
bindInt64(sampleID, to: 1, in: statement)
bindInt64(versionID, to: 2, in: statement)
}
return existing != nil
}
private func insertOpenVisibilityRange(
sampleID: Int64,
versionID: Int64,
observationID: Int64,
observedAt: Date,
db: OpaquePointer?,
statementCache: SQLiteStatementCache? = nil,
allowConflict: Bool = true
) throws {
let insertMode = allowConflict ? "INSERT OR IGNORE" : "INSERT"
try withStatement(
"""
\(insertMode) INTO sample_visibility_ranges (
sample_id, version_id, first_observation_id, last_observation_id, first_seen_at, last_seen_at
) VALUES (?, ?, ?, NULL, ?, NULL)
""",
db: db,
statementCache: statementCache
) { statement in
bindInt64(sampleID, to: 1, in: statement)
bindInt64(versionID, to: 2, in: statement)
bindInt64(observationID, to: 3, in: statement)
sqlite3_bind_double(statement, 4, observedAt.timeIntervalSince1970)
guard sqlite3_step(statement) == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
}
private func rebuildTypeSummary(observationID: Int64, sampleTypeID: Int64, db: OpaquePointer?) throws {
let summary = try typeSummary(observationID: observationID, sampleTypeID: sampleTypeID, db: db)
try upsertTypeSummary(
observationID: observationID,
sampleTypeID: sampleTypeID,
counts: (
appeared: summary.appearedCount,
disappeared: summary.disappearedCount,
representationChanged: summary.representationChangedCount
),
aggregate: ArchiveV2VisibleAggregate(
visibleRecordCount: summary.visibleRecordCount,
earliestStartDate: summary.earliestStartDate,
latestEndDate: summary.latestEndDate,
valueSum: summary.valueSum,
valueMax: summary.valueMax
),
db: db
)
}
private func upsertTypeSummary(
observationID: Int64,
sampleTypeID: Int64,
counts: (appeared: Int, disappeared: Int, representationChanged: Int),
aggregate: ArchiveV2VisibleAggregate,
db: OpaquePointer?
) throws {
let aggregateParts: [String?] = [
String(observationID),
String(sampleTypeID),
String(aggregate.visibleRecordCount),
String(counts.appeared),
String(counts.disappeared),
String(counts.representationChanged),
aggregate.earliestStartDate.map { String($0) },
aggregate.latestEndDate.map { String($0) },
aggregate.valueSum.map { String(format: "%.17g", $0) },
aggregate.valueMax.map { String(format: "%.17g", $0) }
]
let aggregateHash = HashService.archiveContentHash(
domain: "hp:v2:type_summary",
parts: aggregateParts
)
try withStatement(
"""
INSERT OR REPLACE INTO observation_type_summaries (
observation_id, sample_type_id, visible_record_count, appeared_count,
disappeared_count, representation_changed_count, earliest_start_date,
latest_end_date, value_sum, value_max, aggregate_hash
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
db: db
) { statement in
bindInt64(observationID, to: 1, in: statement)
bindInt64(sampleTypeID, to: 2, in: statement)
bindInt(aggregate.visibleRecordCount, to: 3, in: statement)
bindInt(counts.appeared, to: 4, in: statement)
bindInt(counts.disappeared, to: 5, in: statement)
bindInt(counts.representationChanged, to: 6, in: statement)
bindDouble(aggregate.earliestStartDate, to: 7, in: statement)
bindDouble(aggregate.latestEndDate, to: 8, in: statement)
bindDouble(aggregate.valueSum, to: 9, in: statement)
bindDouble(aggregate.valueMax, to: 10, in: statement)
bindText(aggregateHash, to: 11, in: statement)
guard sqlite3_step(statement) == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
}
private func previousTypeSummary(
sampleTypeID: Int64,
beforeObservationID observationID: Int64,
db: OpaquePointer?,
statementCache: SQLiteStatementCache? = nil
) throws -> (observationID: Int64, aggregate: ArchiveV2VisibleAggregate)? {
let sql = """
SELECT observation_id, visible_record_count, earliest_start_date, latest_end_date, value_sum, value_max
FROM observation_type_summaries
WHERE sample_type_id = ? AND observation_id < ?
ORDER BY observation_id DESC
LIMIT 1
"""
return try withStatement(sql, db: db, statementCache: statementCache) { statement in
bindInt64(sampleTypeID, to: 1, in: statement)
bindInt64(observationID, to: 2, in: statement)
guard sqlite3_step(statement) == SQLITE_ROW else {
return nil
}
return (
observationID: sqlite3_column_int64(statement, 0),
aggregate: ArchiveV2VisibleAggregate(
visibleRecordCount: columnInt(statement, 1) ?? 0,
earliestStartDate: columnDouble(statement, 2),
latestEndDate: columnDouble(statement, 3),
valueSum: columnDouble(statement, 4),
valueMax: columnDouble(statement, 5)
)
)
}
}
private func rebuildDailyAggregates(
observationID: Int64,
sampleTypeID: Int64,
observedAt: Date,
db: OpaquePointer?
) throws {
let secondsFromGMT = TimeZone.current.secondsFromGMT(for: observedAt)
try withStatement(
"DELETE FROM daily_type_aggregates WHERE observation_id = ? AND sample_type_id = ?",
db: db
) { statement in
bindInt64(observationID, to: 1, in: statement)
bindInt64(sampleTypeID, to: 2, in: statement)
guard sqlite3_step(statement) == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
let rows = try dailyAggregateRows(sampleTypeID: sampleTypeID, secondsFromGMT: secondsFromGMT, db: db)
try insertDailyAggregateRows(rows, observationID: observationID, sampleTypeID: sampleTypeID, db: db)
}
private func copyDailyAggregates(
fromObservationID: Int64,
toObservationID: Int64,
sampleTypeID: Int64,
db: OpaquePointer?
) throws {
try withStatement(
"DELETE FROM daily_type_aggregates WHERE observation_id = ? AND sample_type_id = ?",
db: db
) { statement in
bindInt64(toObservationID, to: 1, in: statement)
bindInt64(sampleTypeID, to: 2, in: statement)
guard sqlite3_step(statement) == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
let rows = try dailyAggregateRows(
observationID: fromObservationID,
sampleTypeID: sampleTypeID,
db: db
)
try insertDailyAggregateRows(
rows,
observationID: toObservationID,
sampleTypeID: sampleTypeID,
db: db
)
}
private func insertDailyAggregateRows(
_ rows: [ArchiveV2DailyAggregateRow],
observationID: Int64,
sampleTypeID: Int64,
db: OpaquePointer?
) throws {
try withStatement(
"""
INSERT INTO daily_type_aggregates (
observation_id, sample_type_id, bucket_start, bucket_end,
visible_record_count, value_sum, value_max, source_revision_id, aggregate_hash
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
db: db
) { statement in
for row in rows {
sqlite3_reset(statement)
sqlite3_clear_bindings(statement)
let aggregateHash = HashService.archiveContentHash(
domain: "hp:v2:daily_type_aggregate",
parts: row.hashParts(observationID: observationID, sampleTypeID: sampleTypeID)
)
bindInt64(observationID, to: 1, in: statement)
bindInt64(sampleTypeID, to: 2, in: statement)
bindDouble(row.bucketStart, to: 3, in: statement)
bindDouble(row.bucketEnd, to: 4, in: statement)
bindInt(row.visibleRecordCount, to: 5, in: statement)
bindDouble(row.valueSum, to: 6, in: statement)
bindDouble(row.valueMax, to: 7, in: statement)
bindInt64(row.sourceRevisionID, to: 8, in: statement)
bindText(aggregateHash, to: 9, in: statement)
guard sqlite3_step(statement) == SQLITE_DONE else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
}
}
private func dailyAggregateRows(
observationID: Int64,
sampleTypeID: Int64,
db: OpaquePointer?
) throws -> [ArchiveV2DailyAggregateRow] {
let sql = """
SELECT bucket_start, bucket_end, visible_record_count, value_sum, value_max, source_revision_id
FROM daily_type_aggregates
WHERE observation_id = ? AND sample_type_id = ?
ORDER BY bucket_start ASC, source_revision_id ASC
"""
return try withStatement(sql, db: db) { statement in
bindInt64(observationID, to: 1, in: statement)
bindInt64(sampleTypeID, to: 2, in: statement)
var rows: [ArchiveV2DailyAggregateRow] = []
while sqlite3_step(statement) == SQLITE_ROW {
rows.append(ArchiveV2DailyAggregateRow(
bucketStart: sqlite3_column_double(statement, 0),
bucketEnd: sqlite3_column_double(statement, 1),
visibleRecordCount: columnInt(statement, 2) ?? 0,
valueSum: columnDouble(statement, 3),
valueMax: columnDouble(statement, 4),
sourceRevisionID: columnInt64(statement, 5)
))
}
return rows
}
}
private func dailyAggregateRows(
sampleTypeID: Int64,
secondsFromGMT: Int,
db: OpaquePointer?
) throws -> [ArchiveV2DailyAggregateRow] {
let sql = """
WITH typed_samples AS (
SELECT id
FROM samples INDEXED BY idx_samples_type_id
WHERE sample_type_id = ?
)
SELECT
CAST(((v.start_date + ?) / 86400) AS INTEGER) * 86400 - ? AS bucket_start,
CAST(((v.start_date + ?) / 86400) AS INTEGER) * 86400 - ? + 86400 AS bucket_end,
COUNT(*),
SUM(v.numeric_value),
MAX(v.numeric_value),
v.source_revision_id
FROM typed_samples s
JOIN sample_visibility_ranges r INDEXED BY idx_visibility_sample_open_version
ON r.sample_id = s.id
AND r.last_observation_id IS NULL
JOIN sample_versions v ON v.id = r.version_id
GROUP BY bucket_start, bucket_end, v.source_revision_id
ORDER BY bucket_start ASC, v.source_revision_id ASC
"""
return try withStatement(sql, db: db) { statement in
bindInt64(sampleTypeID, to: 1, in: statement)
sqlite3_bind_double(statement, 2, Double(secondsFromGMT))
sqlite3_bind_double(statement, 3, Double(secondsFromGMT))
sqlite3_bind_double(statement, 4, Double(secondsFromGMT))
sqlite3_bind_double(statement, 5, Double(secondsFromGMT))
var rows: [ArchiveV2DailyAggregateRow] = []
while sqlite3_step(statement) == SQLITE_ROW {
rows.append(ArchiveV2DailyAggregateRow(
bucketStart: sqlite3_column_double(statement, 0),
bucketEnd: sqlite3_column_double(statement, 1),
visibleRecordCount: columnInt(statement, 2) ?? 0,
valueSum: columnDouble(statement, 3),
valueMax: columnDouble(statement, 4),
sourceRevisionID: columnInt64(statement, 5)
))
}
return rows
}
}
private func typeSummary(observationID: Int64, sampleTypeID: Int64, db: OpaquePointer?) throws -> ArchiveV2TypeSummary {
let counts = try eventCounts(observationID: observationID, sampleTypeID: sampleTypeID, db: db)
let aggregate = try visibleAggregate(sampleTypeID: sampleTypeID, db: db)
return ArchiveV2TypeSummary(
visibleRecordCount: aggregate.visibleRecordCount,
appearedCount: counts.appeared,
disappearedCount: counts.disappeared,
representationChangedCount: counts.representationChanged,
earliestStartDate: aggregate.earliestStartDate,
latestEndDate: aggregate.latestEndDate,
valueSum: aggregate.valueSum,
valueMax: aggregate.valueMax
)
}
private func eventCounts(
observationID: Int64,
sampleTypeID: Int64,
db: OpaquePointer?
) throws -> (appeared: Int, disappeared: Int, representationChanged: Int) {
let sql = """
SELECT e.event_kind, COUNT(*)
FROM samples s INDEXED BY idx_samples_type_id
JOIN sample_observation_events e INDEXED BY idx_events_sample
ON e.sample_id = s.id
AND e.observation_id = ?
WHERE s.sample_type_id = ?
GROUP BY e.event_kind
"""
return try withStatement(sql, db: db) { statement in
bindInt64(observationID, to: 1, in: statement)
bindInt64(sampleTypeID, to: 2, in: statement)
var appeared = 0
var disappeared = 0
var representationChanged = 0
while sqlite3_step(statement) == SQLITE_ROW {
let kind = columnText(statement, 0)
let count = columnInt(statement, 1) ?? 0
switch kind {
case "appeared":
appeared = count
case "disappeared":
disappeared = count
case "representationChanged":
representationChanged = count
default:
break
}
}
return (appeared, disappeared, representationChanged)
}
}
private func visibleAggregate(sampleTypeID: Int64, db: OpaquePointer?) throws -> ArchiveV2VisibleAggregate {
let sql = """
WITH typed_samples AS (
SELECT id
FROM samples INDEXED BY idx_samples_type_id
WHERE sample_type_id = ?
)
SELECT COUNT(*), MIN(v.start_date), MAX(v.end_date), SUM(v.numeric_value), MAX(v.numeric_value)
FROM typed_samples s
JOIN sample_visibility_ranges r INDEXED BY idx_visibility_sample_open_version
ON r.sample_id = s.id
AND r.last_observation_id IS NULL
JOIN sample_versions v ON v.id = r.version_id
"""
return try withStatement(sql, db: db) { statement in
bindInt64(sampleTypeID, to: 1, in: statement)
guard sqlite3_step(statement) == SQLITE_ROW else {
return ArchiveV2VisibleAggregate(
visibleRecordCount: 0,
earliestStartDate: nil,
latestEndDate: nil,
valueSum: nil,
valueMax: nil
)
}
return ArchiveV2VisibleAggregate(
visibleRecordCount: columnInt(statement, 0) ?? 0,
earliestStartDate: columnDouble(statement, 1),
latestEndDate: columnDouble(statement, 2),
valueSum: columnDouble(statement, 3),
valueMax: columnDouble(statement, 4)
)
}
}
private func bindDiffObservationIDs(
_ fromObservationID: Int64,
_ toObservationID: Int64,
to statement: OpaquePointer?,
startingAt index: inout Int32
) {
bindInt64(fromObservationID, to: index, in: statement)
index += 1
bindInt64(fromObservationID, to: index, in: statement)
index += 1
bindInt64(toObservationID, to: index, in: statement)
index += 1
bindInt64(toObservationID, to: index, in: statement)
index += 1
}
private func archiveRecord(from statement: OpaquePointer?) -> ArchivedHealthRecord {
ArchivedHealthRecord(
id: columnText(statement, 0) ?? "",
sampleTypeIdentifier: columnText(statement, 1) ?? "",
strictFingerprint: columnText(statement, 2) ?? "",
semanticFingerprint: columnText(statement, 3),
healthKitUUIDHash: columnText(statement, 4),
startDate: columnUnixDate(statement, 5) ?? Date(timeIntervalSince1970: 0),
endDate: columnUnixDate(statement, 6) ?? Date(timeIntervalSince1970: 0),
firstSeenAt: columnUnixDate(statement, 7) ?? Date(timeIntervalSince1970: 0),
lastSeenAt: columnUnixDate(statement, 8),
lastVerifiedAt: columnUnixDate(statement, 9),
disappearedAt: columnUnixDate(statement, 10),
valueKind: columnText(statement, 11),
value: columnDouble(statement, 12),
unit: columnText(statement, 13),
categoryValue: columnInt(statement, 14),
workoutActivityType: columnInt(statement, 15),
durationSeconds: columnDouble(statement, 16),
sourceName: nil,
sourceBundleIdentifier: columnText(statement, 17),
deviceName: nil
)
}
private func requiredInt64(
_ sql: String,
db: OpaquePointer?,
statementCache: SQLiteStatementCache? = nil,
bind: (OpaquePointer?) throws -> Void
) throws -> Int64 {
guard let value = try optionalInt64(sql, db: db, statementCache: statementCache, bind: bind) else {
throw SQLiteHealthArchiveStoreError.stepFailed("missing required row")
}
return value
}
private func optionalInt64(
_ sql: String,
db: OpaquePointer?,
statementCache: SQLiteStatementCache? = nil,
bind: (OpaquePointer?) throws -> Void
) throws -> Int64? {
try withStatement(sql, db: db, statementCache: statementCache) { statement in
try bind(statement)
guard sqlite3_step(statement) == SQLITE_ROW else { return nil }
return sqlite3_column_int64(statement, 0)
}
}
private func firstText(_ sql: String, db: OpaquePointer?) throws -> String? {
try withStatement(sql, db: db) { statement in
guard sqlite3_step(statement) == SQLITE_ROW else { return nil }
return columnText(statement, 0)
}
}
private func countRows(_ sql: String, db: OpaquePointer?) throws -> Int {
try withStatement(sql, db: db) { statement in
var count = 0
while sqlite3_step(statement) == SQLITE_ROW {
count += 1
}
return count
}
}
private func execute(_ sql: String, db: OpaquePointer?) throws {
guard sqlite3_exec(db, sql, nil, nil, nil) == SQLITE_OK else {
throw SQLiteHealthArchiveStoreError.stepFailed(lastErrorMessage(db))
}
}
private func withStatement<T>(
_ sql: String,
db: OpaquePointer?,
statementCache: SQLiteStatementCache? = nil,
body: (OpaquePointer?) throws -> T
) throws -> T {
if let statementCache {
return try statementCache.withStatement(sql, body: body)
}
var statement: OpaquePointer?
guard sqlite3_prepare_v2(db, sql, -1, &statement, nil) == SQLITE_OK else {
throw SQLiteHealthArchiveStoreError.prepareFailed(lastErrorMessage(db))
}
defer { sqlite3_finalize(statement) }
return try body(statement)
}
}
private final class SQLiteStatementCache: @unchecked Sendable {
private let db: OpaquePointer?
private var statements: [String: OpaquePointer?] = [:]
nonisolated init(db: OpaquePointer?) {
self.db = db
}
deinit {
finalizeAll()
}
nonisolated(unsafe) func withStatement<T>(_ sql: String, body: (OpaquePointer?) throws -> T) throws -> T {
let statement: OpaquePointer?
if let cached = statements[sql] {
statement = cached
sqlite3_reset(statement)
sqlite3_clear_bindings(statement)
} else {
var prepared: OpaquePointer?
guard sqlite3_prepare_v2(db, sql, -1, &prepared, nil) == SQLITE_OK else {
throw SQLiteHealthArchiveStoreError.prepareFailed(lastErrorMessage(db))
}
statements[sql] = prepared
statement = prepared
}
return try body(statement)
}
nonisolated(unsafe) func finalizeAll() {
for statement in statements.values {
sqlite3_finalize(statement)
}
statements.removeAll()
}
}
private struct SQLiteWriteLookupSourceKey: Hashable, Sendable {
let sourceNameHash: String?
let bundleIdentifier: String?
}
private struct SQLiteWriteLookupSourceRevisionKey: Hashable, Sendable {
let sourceNameHash: String?
let bundleIdentifier: String?
let productType: String?
let version: String?
let operatingSystemVersion: String?
}
private struct SQLiteWriteLookupDeviceKey: Hashable, Sendable {
let deviceHash: String?
let localIdentifierHash: String?
let udiHash: String?
let model: String?
}
private final class SQLiteWriteLookupCache: @unchecked Sendable {
nonisolated init() {}
nonisolated(unsafe) var sampleTypeIDs: [String: Int64] = [:]
nonisolated(unsafe) var sourceIDs: [SQLiteWriteLookupSourceKey: Int64] = [:]
nonisolated(unsafe) var sourceRevisionIDs: [SQLiteWriteLookupSourceRevisionKey: Int64] = [:]
nonisolated(unsafe) var deviceIDs: [SQLiteWriteLookupDeviceKey: Int64] = [:]
nonisolated(unsafe) var metadataBlobIDs: [String: Int64] = [:]
}
private struct ArchiveSampleRow {
let sampleUUIDHash: String
let typeIdentifier: String
let strictFingerprint: String
let semanticFingerprint: String
let payloadHash: String
let startDate: Date
let endDate: Date
let observedAt: Date
let valueKind: String?
let value: Double?
let unit: String?
let categoryValue: Int?
let workoutActivityType: Int?
let durationSeconds: Double?
let sourceName: String?
let sourceBundleIdentifier: String?
let sourceProductType: String?
let sourceVersion: String?
let sourceOperatingSystemVersion: String?
let deviceName: String?
let deviceManufacturer: String?
let deviceModel: String?
let deviceHardwareVersion: String?
let deviceFirmwareVersion: String?
let deviceSoftwareVersion: String?
let deviceLocalIdentifier: String?
let deviceUDI: String?
let metadataJSON: String?
let metadataHash: String?
nonisolated var hasDeviceProvenance: Bool {
deviceName != nil ||
deviceManufacturer != nil ||
deviceModel != nil ||
deviceHardwareVersion != nil ||
deviceFirmwareVersion != nil ||
deviceSoftwareVersion != nil ||
deviceLocalIdentifier != nil ||
deviceUDI != nil
}
nonisolated init(sample: HKSample, observedAt: Date) {
let sampleUUID = sample.uuid.uuidString
let typeIdentifier = sample.sampleType.identifier
let startDate = sample.startDate
let endDate = sample.endDate
let quantity = ArchiveSampleRow.quantityPayload(sample)
let category = sample as? HKCategorySample
let workout = sample as? HKWorkout
let sourceRevision = sample.sourceRevision
let device = sample.device
let valueKind = quantity?.kind ?? (category == nil ? (workout == nil ? nil : "workout") : "category")
let numericValue = quantity?.value
let unit = quantity?.unit
let categoryValue = category?.value
let workoutActivityType = workout.map { Int($0.workoutActivityType.rawValue) }
let durationSeconds = workout?.duration
let sourceBundleIdentifier = sourceRevision.source.bundleIdentifier
let sourceProductType = sourceRevision.productType
let sourceVersion = sourceRevision.version
let sourceOperatingSystemVersion = ArchiveSampleRow.operatingSystemVersionString(sourceRevision.operatingSystemVersion)
let deviceModel = device?.model
let deviceHardwareVersion = device?.hardwareVersion
let deviceFirmwareVersion = device?.firmwareVersion
let deviceSoftwareVersion = device?.softwareVersion
let deviceLocalIdentifier = device?.localIdentifier
let deviceUDI = device?.udiDeviceIdentifier
let metadataJSON = ArchiveSampleRow.metadataJSONString(sample.metadata)
let metadataHash = metadataJSON.map { HashService.archiveContentHash(domain: "hp:v2:metadata", parts: [$0]) }
self.sampleUUIDHash = HashService.sampleUUIDHash(sampleUUID)
self.typeIdentifier = typeIdentifier
self.strictFingerprint = HashService.sampleFingerprint(
typeIdentifier: typeIdentifier,
sampleUUID: sampleUUID,
startDate: startDate,
endDate: endDate
)
self.semanticFingerprint = HashService.archiveSemanticFingerprint(
typeIdentifier: typeIdentifier,
startDate: startDate,
endDate: endDate,
value: numericValue,
unit: unit,
categoryValue: categoryValue,
workoutActivityType: workout?.workoutActivityType.rawValue,
sourceBundleIdentifier: sourceBundleIdentifier
)
self.valueKind = valueKind
self.value = numericValue
self.unit = unit
self.categoryValue = categoryValue
self.workoutActivityType = workoutActivityType
self.durationSeconds = durationSeconds
self.sourceName = sourceRevision.source.name
self.sourceBundleIdentifier = sourceBundleIdentifier
self.sourceProductType = sourceProductType
self.sourceVersion = sourceVersion
self.sourceOperatingSystemVersion = sourceOperatingSystemVersion
self.deviceName = device?.name
self.deviceManufacturer = device?.manufacturer
self.deviceModel = deviceModel
self.deviceHardwareVersion = deviceHardwareVersion
self.deviceFirmwareVersion = deviceFirmwareVersion
self.deviceSoftwareVersion = deviceSoftwareVersion
self.deviceLocalIdentifier = deviceLocalIdentifier
self.deviceUDI = deviceUDI
self.metadataJSON = metadataJSON
self.metadataHash = metadataHash
self.payloadHash = HashService.archiveContentHash(
domain: "hp:v2:payload",
parts: [
typeIdentifier,
ArchiveSampleRow.timestampString(startDate),
ArchiveSampleRow.timestampString(endDate),
valueKind,
numericValue.map { String(format: "%.17g", $0) },
unit,
categoryValue.map(String.init),
workoutActivityType.map(String.init),
durationSeconds.map { String(format: "%.17g", $0) },
sourceBundleIdentifier,
sourceProductType,
sourceVersion,
sourceOperatingSystemVersion,
deviceModel,
deviceHardwareVersion,
deviceFirmwareVersion,
deviceSoftwareVersion,
deviceLocalIdentifier,
deviceUDI,
metadataHash
]
)
self.startDate = startDate
self.endDate = endDate
self.observedAt = observedAt
}
nonisolated private static func timestampString(_ date: Date) -> String {
String(format: "%.6f", date.timeIntervalSince1970)
}
nonisolated private static func quantityPayload(_ sample: HKSample) -> (kind: String, value: Double?, unit: String?)? {
guard let sample = sample as? HKQuantitySample else { return nil }
let identifier = sample.quantityType.identifier
guard let storageUnit = quantityStorageUnit(for: identifier) else {
return ("quantity", nil, nil)
}
return ("quantity", sample.quantity.doubleValue(for: storageUnit.unit), storageUnit.label)
}
nonisolated private static func quantityStorageUnit(for identifier: String) -> (unit: HKUnit, label: String)? {
switch identifier {
case HKQuantityTypeIdentifier.heartRate.rawValue,
HKQuantityTypeIdentifier.restingHeartRate.rawValue,
"HKQuantityTypeIdentifierWalkingHeartRateAverage":
return (HKUnit.count().unitDivided(by: .minute()), "count/min")
case HKQuantityTypeIdentifier.respiratoryRate.rawValue:
return (HKUnit.count().unitDivided(by: .minute()), "count/min")
case HKQuantityTypeIdentifier.activeEnergyBurned.rawValue,
"HKQuantityTypeIdentifierBasalEnergyBurned",
"HKQuantityTypeIdentifierDietaryEnergyConsumed":
return (.kilocalorie(), "kcal")
case HKQuantityTypeIdentifier.distanceWalkingRunning.rawValue,
"HKQuantityTypeIdentifierDistanceCycling",
"HKQuantityTypeIdentifierDistanceSwimming",
"HKQuantityTypeIdentifierDistanceDownhillSnowSports",
"HKQuantityTypeIdentifierDistanceWheelchair",
"HKQuantityTypeIdentifierHeight",
"HKQuantityTypeIdentifierWaistCircumference",
"HKQuantityTypeIdentifierWalkingStepLength",
"HKQuantityTypeIdentifierSixMinuteWalkTestDistance":
return (.meter(), "m")
case "HKQuantityTypeIdentifierWalkingSpeed",
"HKQuantityTypeIdentifierStairAscentSpeed",
"HKQuantityTypeIdentifierStairDescentSpeed":
return (.meter().unitDivided(by: .second()), "m/s")
case HKQuantityTypeIdentifier.appleExerciseTime.rawValue,
"HKQuantityTypeIdentifierAppleMoveTime",
"HKQuantityTypeIdentifierAppleStandTime":
return (.minute(), "min")
case HKQuantityTypeIdentifier.environmentalAudioExposure.rawValue,
HKQuantityTypeIdentifier.headphoneAudioExposure.rawValue:
return (.decibelAWeightedSoundPressureLevel(), "dBASPL")
case HKQuantityTypeIdentifier.bodyMass.rawValue,
"HKQuantityTypeIdentifierLeanBodyMass":
return (HKUnit.gramUnit(with: .kilo), "kg")
case HKQuantityTypeIdentifier.vo2Max.rawValue:
return (
HKUnit.literUnit(with: .milli)
.unitDivided(by: HKUnit.gramUnit(with: .kilo))
.unitDivided(by: .minute()),
"mL/kg/min"
)
case HKQuantityTypeIdentifier.stepCount.rawValue,
"HKQuantityTypeIdentifierFlightsClimbed",
"HKQuantityTypeIdentifierSwimmingStrokeCount",
"HKQuantityTypeIdentifierPushCount",
"HKQuantityTypeIdentifierInhalerUsage",
"HKQuantityTypeIdentifierBodyMassIndex":
return (.count(), "count")
case "HKQuantityTypeIdentifierHeartRateVariabilitySDNN":
return (HKUnit.secondUnit(with: .milli), "ms")
case "HKQuantityTypeIdentifierAtrialFibrillationBurden",
"HKQuantityTypeIdentifierOxygenSaturation",
"HKQuantityTypeIdentifierBodyFatPercentage",
"HKQuantityTypeIdentifierPeripheralPerfusionIndex",
"HKQuantityTypeIdentifierWalkingAsymmetryPercentage",
"HKQuantityTypeIdentifierWalkingDoubleSupportPercentage",
"HKQuantityTypeIdentifierAppleWalkingSteadiness":
return (.percent(), "%")
case "HKQuantityTypeIdentifierForcedVitalCapacity",
"HKQuantityTypeIdentifierForcedExpiratoryVolume1":
return (.liter(), "L")
case "HKQuantityTypeIdentifierPeakExpiratoryFlowRate":
return (.liter().unitDivided(by: .minute()), "L/min")
case "HKQuantityTypeIdentifierBodyTemperature",
"HKQuantityTypeIdentifierBasalBodyTemperature":
return (.degreeCelsius(), "degC")
case "HKQuantityTypeIdentifierBloodPressureSystolic",
"HKQuantityTypeIdentifierBloodPressureDiastolic":
return (.millimeterOfMercury(), "mmHg")
case "HKQuantityTypeIdentifierBloodGlucose":
return (HKUnit.gramUnit(with: .milli).unitDivided(by: .literUnit(with: .deci)), "mg/dL")
case "HKQuantityTypeIdentifierInsulinDelivery":
return (.internationalUnit(), "IU")
case "HKQuantityTypeIdentifierElectrodermalActivity":
return (HKUnit.siemenUnit(with: .micro), "uS")
case "HKQuantityTypeIdentifierDietaryWater":
return (HKUnit.literUnit(with: .milli), "mL")
case "HKQuantityTypeIdentifierDietaryProtein",
"HKQuantityTypeIdentifierDietaryCarbohydrates",
"HKQuantityTypeIdentifierDietaryFiber",
"HKQuantityTypeIdentifierDietarySugar",
"HKQuantityTypeIdentifierDietaryFatTotal",
"HKQuantityTypeIdentifierDietaryFatSaturated",
"HKQuantityTypeIdentifierDietaryFatMonounsaturated",
"HKQuantityTypeIdentifierDietaryFatPolyunsaturated",
"HKQuantityTypeIdentifierDietaryCholesterol",
"HKQuantityTypeIdentifierDietarySodium",
"HKQuantityTypeIdentifierDietaryPotassium",
"HKQuantityTypeIdentifierDietaryCalcium",
"HKQuantityTypeIdentifierDietaryIron",
"HKQuantityTypeIdentifierDietaryMagnesium",
"HKQuantityTypeIdentifierDietaryZinc",
"HKQuantityTypeIdentifierDietaryVitaminA",
"HKQuantityTypeIdentifierDietaryVitaminB6",
"HKQuantityTypeIdentifierDietaryVitaminB12",
"HKQuantityTypeIdentifierDietaryVitaminC",
"HKQuantityTypeIdentifierDietaryVitaminD",
"HKQuantityTypeIdentifierDietaryVitaminE",
"HKQuantityTypeIdentifierDietaryVitaminK",
"HKQuantityTypeIdentifierDietaryCaffeine":
return (.gram(), "g")
default:
return nil
}
}
nonisolated private static func metadataJSONString(_ metadata: [String: Any]?) -> String? {
guard let metadata, !metadata.isEmpty else { return nil }
let sanitized = metadata.mapValues(sanitize)
guard JSONSerialization.isValidJSONObject(sanitized),
let data = try? JSONSerialization.data(withJSONObject: sanitized, options: [.sortedKeys]) else {
return nil
}
return String(data: data, encoding: .utf8)
}
nonisolated private static func sanitize(_ value: Any) -> Any {
switch value {
case let value as String:
return value
case let value as NSNumber:
return value
case let value as Date:
return ISO8601DateFormatter().string(from: value)
case let value as UUID:
return value.uuidString
case let values as [Any]:
return values.map(sanitize)
case let values as [String: Any]:
return values.mapValues(sanitize)
default:
return String(describing: value)
}
}
nonisolated private static func operatingSystemVersionString(_ version: OperatingSystemVersion) -> String? {
guard version.majorVersion != 0 || version.minorVersion != 0 || version.patchVersion != 0 else {
return nil
}
guard (0...100).contains(version.majorVersion),
(0...1_000).contains(version.minorVersion),
(0...1_000).contains(version.patchVersion) else {
return nil
}
return "\(version.majorVersion).\(version.minorVersion).\(version.patchVersion)"
}
}
private enum ArchiveV2SampleWriteKind {
case inserted
case updated
case unchanged
}
private struct ArchiveV2SampleWriteResult {
let sampleTypeID: Int64
let kind: ArchiveV2SampleWriteKind
}
private struct ArchiveV2TypeSummary {
let visibleRecordCount: Int
let appearedCount: Int
let disappearedCount: Int
let representationChangedCount: Int
let earliestStartDate: Double?
let latestEndDate: Double?
let valueSum: Double?
let valueMax: Double?
}
private struct ArchiveV2VisibleAggregate {
let visibleRecordCount: Int
let earliestStartDate: Double?
let latestEndDate: Double?
let valueSum: Double?
let valueMax: Double?
}
private struct ArchiveV2DailyAggregateRow {
let bucketStart: Double
let bucketEnd: Double
let visibleRecordCount: Int
let valueSum: Double?
let valueMax: Double?
let sourceRevisionID: Int64?
nonisolated func hashParts(observationID: Int64, sampleTypeID: Int64) -> [String?] {
[
String(observationID),
String(sampleTypeID),
String(bucketStart),
String(bucketEnd),
String(visibleRecordCount),
valueSum.map { String(format: "%.17g", $0) },
valueMax.map { String(format: "%.17g", $0) },
sourceRevisionID.map(String.init)
]
}
}
nonisolated private struct ExportRecordPayload: Encodable {
let id: String
let sampleTypeIdentifier: String
let strictFingerprint: String
let semanticFingerprint: String?
let healthKitUUIDHash: String?
let startDate: Date
let endDate: Date
let firstSeenAt: Date
let lastSeenAt: Date?
let lastVerifiedAt: Date?
let disappearedAt: Date?
let valueKind: String?
let value: Double?
let unit: String?
let categoryValue: Int?
let workoutActivityType: Int?
let durationSeconds: Double?
let sourceName: String?
let sourceBundleIdentifier: String?
let deviceName: String?
let displayValue: String?
init(_ record: ArchivedHealthRecord) {
id = record.id
sampleTypeIdentifier = record.sampleTypeIdentifier
strictFingerprint = record.strictFingerprint
semanticFingerprint = record.semanticFingerprint
healthKitUUIDHash = record.healthKitUUIDHash
startDate = record.startDate
endDate = record.endDate
firstSeenAt = record.firstSeenAt
lastSeenAt = record.lastSeenAt
lastVerifiedAt = record.lastVerifiedAt
disappearedAt = record.disappearedAt
valueKind = record.valueKind
value = record.value
unit = record.unit
categoryValue = record.categoryValue
workoutActivityType = record.workoutActivityType
durationSeconds = record.durationSeconds
sourceName = record.sourceName
sourceBundleIdentifier = record.sourceBundleIdentifier
deviceName = record.deviceName
displayValue = record.displayValue
}
}
nonisolated private extension JSONEncoder {
static var healthArchive: JSONEncoder {
let encoder = JSONEncoder()
encoder.dateEncodingStrategy = .iso8601
encoder.outputFormatting = [.prettyPrinted, .sortedKeys]
return encoder
}
static func healthArchiveDateString(_ date: Date) -> String {
let data = try? healthArchive.encode(date)
let encoded = data.map { String(decoding: $0, as: UTF8.self) } ?? "\"\(date.timeIntervalSince1970)\""
return encoded.trimmingCharacters(in: CharacterSet(charactersIn: "\""))
}
}
nonisolated private func bindText(_ value: String?, to index: Int32, in statement: OpaquePointer?) {
guard let value else {
sqlite3_bind_null(statement, index)
return
}
sqlite3_bind_text(statement, index, value, -1, unsafeBitCast(-1, to: sqlite3_destructor_type.self))
}
nonisolated private func bindData(_ value: Data?, to index: Int32, in statement: OpaquePointer?) {
guard let value else {
sqlite3_bind_null(statement, index)
return
}
value.withUnsafeBytes { rawBuffer in
let pointer = rawBuffer.baseAddress
sqlite3_bind_blob(statement, index, pointer, Int32(value.count), unsafeBitCast(-1, to: sqlite3_destructor_type.self))
}
}
nonisolated private func bindDouble(_ value: Double?, to index: Int32, in statement: OpaquePointer?) {
guard let value else {
sqlite3_bind_null(statement, index)
return
}
sqlite3_bind_double(statement, index, value)
}
nonisolated private func bindInt(_ value: Int?, to index: Int32, in statement: OpaquePointer?) {
guard let value else {
sqlite3_bind_null(statement, index)
return
}
sqlite3_bind_int64(statement, index, sqlite3_int64(value))
}
nonisolated private func bindInt64(_ value: Int64?, to index: Int32, in statement: OpaquePointer?) {
guard let value else {
sqlite3_bind_null(statement, index)
return
}
sqlite3_bind_int64(statement, index, sqlite3_int64(value))
}
nonisolated private func columnText(_ statement: OpaquePointer?, _ index: Int32) -> String? {
guard sqlite3_column_type(statement, index) != SQLITE_NULL,
let pointer = sqlite3_column_text(statement, index) else {
return nil
}
return String(cString: pointer)
}
nonisolated private func columnData(_ statement: OpaquePointer?, _ index: Int32) -> Data? {
guard sqlite3_column_type(statement, index) != SQLITE_NULL,
let bytes = sqlite3_column_blob(statement, index) else {
return nil
}
let count = Int(sqlite3_column_bytes(statement, index))
return Data(bytes: bytes, count: count)
}
nonisolated private func columnDate(_ statement: OpaquePointer?, _ index: Int32) -> Date? {
guard sqlite3_column_type(statement, index) != SQLITE_NULL else { return nil }
return Date(timeIntervalSinceReferenceDate: sqlite3_column_double(statement, index))
}
nonisolated private func columnUnixDate(_ statement: OpaquePointer?, _ index: Int32) -> Date? {
guard sqlite3_column_type(statement, index) != SQLITE_NULL else { return nil }
return Date(timeIntervalSince1970: sqlite3_column_double(statement, index))
}
nonisolated private func columnDouble(_ statement: OpaquePointer?, _ index: Int32) -> Double? {
guard sqlite3_column_type(statement, index) != SQLITE_NULL else { return nil }
return sqlite3_column_double(statement, index)
}
nonisolated private func columnInt(_ statement: OpaquePointer?, _ index: Int32) -> Int? {
guard sqlite3_column_type(statement, index) != SQLITE_NULL else { return nil }
return Int(sqlite3_column_int(statement, index))
}
nonisolated private func columnInt64(_ statement: OpaquePointer?, _ index: Int32) -> Int64? {
guard sqlite3_column_type(statement, index) != SQLITE_NULL else { return nil }
return sqlite3_column_int64(statement, index)
}
nonisolated private func quotedIdentifier(_ value: String) -> String {
"\"\(value.replacingOccurrences(of: "\"", with: "\"\""))\""
}
nonisolated private func lastErrorMessage(_ db: OpaquePointer?) -> String {
guard let message = sqlite3_errmsg(db) else { return "unknown SQLite error" }
return String(cString: message)
}