Populate daily archive aggregates and integrity checks · 08a5014

Populate daily archive aggregates and integrity checks
Browse files

bogdan committed 2 weeks ago

main

1 parent 8f40c05

commit 08a5014

Showing 4 changed files with 215 additions and 3 deletions

+2 -2

HealthProbe/Doc/04-project/IMPLEMENTATION_STATUS.md

@@ -25,7 +25,7 @@ There are no real deployments, only test installations. Existing prototype datab
 |------|----------------|--------------------|
 | Product docs | Updated | Keep `HealthProbe/Doc/README.md` as canonical index |
 | HealthKit capture | Prototype exists | Adapt capture to write differential SQLite observations first |
-| SQLite archive | Archive v2 schema and differential write path partially implemented; legacy read table still active | Add daily aggregates, SQL reads, integrity tests, then retire `archive_samples` |
+| SQLite archive | Archive v2 schema, differential write path, daily aggregate rebuilds, and integrity report are partially implemented; legacy read table still active | Add SQL reads, open/schema-version tests, idempotency tests, then retire `archive_samples` |
 | Core Data cache | Not implemented | Add rebuildable cache for expensive counts, summaries, report metadata, UI state |
 | SwiftData cache | Exists | Treat as disposable prototype data; reset/ignore during v2 transition |
 | UI | Prototype exists | Reframe screens around observations, diffs, export, archive status |
@@ -38,7 +38,7 @@ There are no real deployments, only test installations. Existing prototype datab
 
 Detailed checkable milestones live in [`Refactoring-Plan.md`](Refactoring-Plan.md).
 
-1. Finish differential write path: daily aggregates and stricter retry/idempotency tests.
+1. Finish differential write path hardening with stricter retry/idempotency tests.
 2. Add SQLite integrity/open/schema-version tests.
 3. Move archive reads from `archive_samples` to SQL over visibility ranges and sample versions.
 4. Move large diffs/counts into SQL queries with indexes/temp tables/paged results.


+2 -1

HealthProbe/Doc/04-project/Refactoring-Plan.md

View

@@ -102,6 +102,7 @@ Checklist:
 - [x] Implement `export_manifests`.
 - [x] Implement `export_items`.
 - [x] Add required indexes.
+- [x] Add archive integrity report for schema version, required tables, `PRAGMA integrity_check`, and foreign keys.
 - [ ] Add SQLite integrity/open/schema-version tests.
 
 Acceptance:
@@ -126,7 +127,7 @@ Checklist:
 - [x] Record `HKDeletedObject` evidence by UUID hash.
 - [x] Close visibility ranges for disappeared/deleted samples.
 - [x] Maintain open visibility ranges for visible samples.
-- [ ] Rebuild/update affected aggregates after capture.
+- [x] Rebuild/update affected type summaries and daily aggregates after capture/delete observations.
 - [x] Commit SQLite before Core Data/cache work.
 - [ ] Make repeated capture page writes idempotent.
 


+16 -0

HealthProbe/Services/Protocols/HealthArchiveStore.swift

View

@@ -8,6 +8,7 @@ protocol HealthArchiveStore {
     func recordDisappearance(sampleUUIDHash: String, sampleTypeIdentifier: String, observedMissingAt: Date) async throws
     func records(for request: HealthArchiveRecordRequest) async throws -> [ArchivedHealthRecord]
     func exportReport(_ request: HealthArchiveReportRequest) async throws -> URL
+    func checkIntegrity() async throws -> HealthArchiveIntegrityReport
 }
 
 struct HealthArchiveWriteSummary: Equatable, Sendable {
@@ -16,6 +17,21 @@ struct HealthArchiveWriteSummary: Equatable, Sendable {
     let unchangedCount: Int
 }
 
+struct HealthArchiveIntegrityReport: Equatable, Sendable {
+    let schemaVersion: Int?
+    let sqliteIntegrityStatus: String
+    let foreignKeyIssueCount: Int
+    let requiredTableNames: Set<String>
+    let missingTableNames: Set<String>
+
+    var passed: Bool {
+        schemaVersion == 2 &&
+        sqliteIntegrityStatus == "ok" &&
+        foreignKeyIssueCount == 0 &&
+        missingTableNames.isEmpty
+    }
+}
+
 struct HealthArchiveRecordRequest: Equatable, Sendable {
     let sampleTypeIdentifier: String?
     let fingerprints: Set<String>


+195 -0

HealthProbe/Services/SQLiteHealthArchiveStore.swift

View

+                     actor SQLiteHealthArchiveStore: HealthArchiveStore {
+                         static let shared = SQLiteHealthArchiveStore()
+                         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",
+                    +        "daily_type_aggregates",
+                    +        "export_manifests",
+                    +        "export_items"
+                    +    ]
+                         private let databaseURL: URL
+                         private var didPrepareSchema = false
+                                         db: db
+                                     try rebuildTypeSummary(observationID: observationID, sampleTypeID: sampleTypeID, db: db)
+                    +                try rebuildDailyAggregates(
+                    +                    observationID: observationID,
+                    +                    sampleTypeID: sampleTypeID,
+                    +                    observedAt: observedMissingAt,
+                    +                    db: db
+                    +                )
+                                 let sql = """
+                             return exportURL
+                    +    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(),
+                             for sampleTypeID in touchedTypeIDs {
+                                 try rebuildTypeSummary(observationID: observationID, sampleTypeID: sampleTypeID, db: db)
+                    +            try rebuildDailyAggregates(
+                    +                observationID: observationID,
+                    +                sampleTypeID: sampleTypeID,
+                    +                observedAt: observedAt,
+                    +                db: db
+                    +            )
+                             return HealthArchiveWriteSummary(
+                    +    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 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(
+                    +        sampleTypeID: Int64,
+                    +        secondsFromGMT: Int,
+                    +        db: OpaquePointer?
+                    +    ) throws -> [ArchiveV2DailyAggregateRow] {
+                    +        let sql = """
+                    +        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 sample_visibility_ranges r
+                    +        JOIN samples s ON s.id = r.sample_id
+                    +        JOIN sample_versions v ON v.id = r.version_id
+                    +        WHERE s.sample_type_id = ? AND r.last_observation_id IS NULL
+                    +        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
+                    +            sqlite3_bind_double(statement, 1, Double(secondsFromGMT))
+                    +            sqlite3_bind_double(statement, 2, Double(secondsFromGMT))
+                    +            sqlite3_bind_double(statement, 3, Double(secondsFromGMT))
+                    +            sqlite3_bind_double(statement, 4, Double(secondsFromGMT))
+                    +            bindInt64(sampleTypeID, to: 5, 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 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)
+                    +    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))
+                         let valueMax: Double?
+                    +private struct ArchiveV2DailyAggregateRow {
+                    +    let bucketStart: Double
+                    +    let bucketEnd: Double
+                    +    let visibleRecordCount: Int
+                    +    let valueSum: Double?
+                    +    let valueMax: Double?
+                    +    let sourceRevisionID: Int64?
+                    +    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 HealthArchiveReportPayload: Encodable {
+                         let reportID: UUID
+                         let title: String
+                         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: "\"\""))\""

	@@ -102,6 +102,7 @@ Checklist:
102	102	- [x] Implement `export_manifests`.
103	103	- [x] Implement `export_items`.
104	104	- [x] Add required indexes.
	105	+- [x] Add archive integrity report for schema version, required tables, `PRAGMA integrity_check`, and foreign keys.
105	106	- [ ] Add SQLite integrity/open/schema-version tests.
106	107
107	108	Acceptance:
	@@ -126,7 +127,7 @@ Checklist:
126	127	- [x] Record `HKDeletedObject` evidence by UUID hash.
127	128	- [x] Close visibility ranges for disappeared/deleted samples.
128	129	- [x] Maintain open visibility ranges for visible samples.
129		-- [ ] Rebuild/update affected aggregates after capture.
	130	+- [x] Rebuild/update affected type summaries and daily aggregates after capture/delete observations.
130	131	- [x] Commit SQLite before Core Data/cache work.
131	132	- [ ] Make repeated capture page writes idempotent.
132	133

	@@ -25,7 +25,7 @@ There are no real deployments, only test installations. Existing prototype datab
25	25	\|------\|----------------\|--------------------\|
26	26	\| Product docs \| Updated \| Keep `HealthProbe/Doc/README.md` as canonical index \|
27	27	\| HealthKit capture \| Prototype exists \| Adapt capture to write differential SQLite observations first \|
28		-\| SQLite archive \| Archive v2 schema and differential write path partially implemented; legacy read table still active \| Add daily aggregates, SQL reads, integrity tests, then retire `archive_samples` \|
	28	+\| SQLite archive \| Archive v2 schema, differential write path, daily aggregate rebuilds, and integrity report are partially implemented; legacy read table still active \| Add SQL reads, open/schema-version tests, idempotency tests, then retire `archive_samples` \|
29	29	\| Core Data cache \| Not implemented \| Add rebuildable cache for expensive counts, summaries, report metadata, UI state \|
30	30	\| SwiftData cache \| Exists \| Treat as disposable prototype data; reset/ignore during v2 transition \|
31	31	\| UI \| Prototype exists \| Reframe screens around observations, diffs, export, archive status \|
	@@ -38,7 +38,7 @@ There are no real deployments, only test installations. Existing prototype datab
38	38
39	39	Detailed checkable milestones live in [`Refactoring-Plan.md`](Refactoring-Plan.md).
40	40
41		-1. Finish differential write path: daily aggregates and stricter retry/idempotency tests.
	41	+1. Finish differential write path hardening with stricter retry/idempotency tests.
42	42	2. Add SQLite integrity/open/schema-version tests.
43	43	3. Move archive reads from `archive_samples` to SQL over visibility ranges and sample versions.
44	44	4. Move large diffs/counts into SQL queries with indexes/temp tables/paged results.

	@@ -8,6 +8,7 @@ protocol HealthArchiveStore {
8	8	func recordDisappearance(sampleUUIDHash: String, sampleTypeIdentifier: String, observedMissingAt: Date) async throws
9	9	func records(for request: HealthArchiveRecordRequest) async throws -> [ArchivedHealthRecord]
10	10	func exportReport(_ request: HealthArchiveReportRequest) async throws -> URL
	11	+ func checkIntegrity() async throws -> HealthArchiveIntegrityReport
11	12	}
12	13
13	14	struct HealthArchiveWriteSummary: Equatable, Sendable {
	@@ -16,6 +17,21 @@ struct HealthArchiveWriteSummary: Equatable, Sendable {
16	17	let unchangedCount: Int
17	18	}
18	19
	20	+struct HealthArchiveIntegrityReport: Equatable, Sendable {
	21	+ let schemaVersion: Int?
	22	+ let sqliteIntegrityStatus: String
	23	+ let foreignKeyIssueCount: Int
	24	+ let requiredTableNames: Set<String>
	25	+ let missingTableNames: Set<String>
	26	+
	27	+ var passed: Bool {
	28	+ schemaVersion == 2 &&
	29	+ sqliteIntegrityStatus == "ok" &&
	30	+ foreignKeyIssueCount == 0 &&
	31	+ missingTableNames.isEmpty
	32	+ }
	33	+}
	34	+
19	35	struct HealthArchiveRecordRequest: Equatable, Sendable {
20	36	let sampleTypeIdentifier: String?
21	37	let fingerprints: Set<String>