Skip to content

Commit 1f152d9

Browse files
cameronvoellcameronvoell
committed
start writing to new profile table
1 parent 7969469 commit 1f152d9

7 files changed

Lines changed: 233 additions & 9 deletions

File tree

‎ConvosCore/Sources/ConvosCore/Messaging/MessagingService.swift‎

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -162,6 +162,66 @@ final class MessagingService: MessagingServiceProtocol, @unchecked Sendable {
162162
MyGlobalProfileRepository(sessionStateManager: sessionStateManager, databaseReader: databaseReader)
163163
}
164164

165+
// MARK: Profiles (canonical Profile table, transition)
166+
167+
private lazy var profileStore: any ProfileStoreProtocol = GRDBProfileStore(
168+
databaseWriter: databaseWriter,
169+
databaseReader: databaseReader
170+
)
171+
172+
private lazy var selfProfileStore: any SelfProfileStoreProtocol = GRDBSelfProfileStore(
173+
databaseWriter: databaseWriter,
174+
databaseReader: databaseReader
175+
)
176+
177+
private lazy var sharedProfilesRepository: ProfilesRepository = ProfilesRepository(
178+
profileStore: profileStore,
179+
selfProfileStore: selfProfileStore,
180+
selfInboxIdProvider: { [sessionStateManager] in
181+
(try? await sessionStateManager.waitForInboxReadyResult())?.client.inboxId
182+
}
183+
)
184+
185+
private lazy var profileMemberMirror: ProfileMemberMirror = ProfileMemberMirror(
186+
databaseReader: databaseReader,
187+
profileStore: profileStore,
188+
selfProfileStore: selfProfileStore,
189+
selfInboxIdProvider: { [sessionStateManager] in
190+
(try? await sessionStateManager.waitForInboxReadyResult())?.client.inboxId
191+
}
192+
)
193+
194+
/// Canonical identity source. Dormant until the cutover - nothing renders
195+
/// from it yet.
196+
func profilesRepository() -> ProfilesRepository {
197+
sharedProfilesRepository
198+
}
199+
200+
/// Seeds the canonical stores from legacy `memberProfile` (one-time
201+
/// backfill), warms the repository cache, and starts mirroring ongoing
202+
/// `memberProfile` writes. Self-guards on inbox-ready, so it is safe to call
203+
/// early. The new tables stay dormant until the cutover.
204+
func startProfileMirroring() async {
205+
if let selfInboxId = (try? await sessionStateManager.waitForInboxReadyResult())?.client.inboxId {
206+
do {
207+
try await ProfileBackfill(
208+
databaseReader: databaseReader,
209+
profileStore: profileStore,
210+
selfProfileStore: selfProfileStore,
211+
selfInboxId: selfInboxId
212+
).run()
213+
} catch {
214+
Log.error("Profile backfill failed: \(error)")
215+
}
216+
}
217+
await sharedProfilesRepository.warmUp()
218+
await profileMemberMirror.start()
219+
}
220+
221+
func stopProfileMirroring() async {
222+
await profileMemberMirror.stop()
223+
}
224+
165225
// MARK: New Conversation
166226

167227
func conversationStateManager() -> any ConversationStateManagerProtocol {

‎ConvosCore/Sources/ConvosCore/Profiles/Repository/ProfileBackfill.swift‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,14 @@ struct ProfileBackfill {
3838
let rows = try await databaseReader.read { db in
3939
try DBMemberProfile.fetchAll(db)
4040
}
41+
try await mirror(rows)
42+
}
43+
44+
/// Mirrors the given legacy rows into the canonical stores. Idempotent: a
45+
/// write only happens when the merged value differs from what's stored, so
46+
/// this can be re-run on every `memberProfile` change (see
47+
/// `ProfileMemberMirror`) without churning writes for unchanged rows.
48+
func mirror(_ rows: [DBMemberProfile]) async throws {
4149
guard !rows.isEmpty else { return }
4250

4351
// Collapse a person's multiple conversation rows into one identity, and
@@ -89,7 +97,7 @@ struct ProfileBackfill {
8997
source: .contact,
9098
sentAt: floor
9199
)
92-
if let merged {
100+
if let merged, merged != existing {
93101
try await profileStore.saveAvatar(merged)
94102
}
95103
}
@@ -106,7 +114,9 @@ struct ProfileBackfill {
106114
source: .contact,
107115
sentAt: floor
108116
)
109-
try await profileStore.saveIdentity(merged)
117+
if merged != existing {
118+
try await profileStore.saveIdentity(merged)
119+
}
110120
}
111121
}
112122
}

‎ConvosCore/Sources/ConvosCore/Profiles/Repository/ProfileMemberMirror.swift‎

Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,86 @@
1+
import Foundation
2+
import GRDB
3+
4+
/// Transition bridge that keeps the canonical profile stores in sync with the
5+
/// legacy per-conversation `memberProfile` table, before the direct sync seam
6+
/// lands at cutover. Observes `memberProfile` and re-runs the idempotent
7+
/// backfill mirror on each change, so `DBProfile` / `DBProfileAvatar` track the
8+
/// resolved identity the app currently renders.
9+
///
10+
/// This is intentionally temporary: once `StreamProcessor` writes profile events
11+
/// directly and `memberProfile` is removed, the mirror is deleted.
12+
actor ProfileMemberMirror {
13+
private let databaseReader: any DatabaseReader
14+
private let profileStore: any ProfileStoreProtocol
15+
private let selfProfileStore: any SelfProfileStoreProtocol
16+
private let selfInboxIdProvider: @Sendable () async -> String?
17+
18+
private var observationTask: Task<Void, Never>?
19+
private var cachedSelfInboxId: String?
20+
21+
init(
22+
databaseReader: any DatabaseReader,
23+
profileStore: any ProfileStoreProtocol,
24+
selfProfileStore: any SelfProfileStoreProtocol,
25+
selfInboxIdProvider: @escaping @Sendable () async -> String?
26+
) {
27+
self.databaseReader = databaseReader
28+
self.profileStore = profileStore
29+
self.selfProfileStore = selfProfileStore
30+
self.selfInboxIdProvider = selfInboxIdProvider
31+
}
32+
33+
func start() {
34+
guard observationTask == nil else { return }
35+
observationTask = Task { [weak self] in
36+
await self?.observe()
37+
}
38+
}
39+
40+
func stop() {
41+
observationTask?.cancel()
42+
observationTask = nil
43+
}
44+
45+
private func observe() async {
46+
let dbReader = databaseReader
47+
let stream = ValueObservation
48+
.tracking { db in
49+
try DBMemberProfile.fetchAll(db)
50+
}
51+
// Only re-mirror when the row set actually changes; an unrelated
52+
// write to the same tables can re-emit an identical snapshot.
53+
.removeDuplicates()
54+
.values(in: dbReader)
55+
do {
56+
for try await rows in stream {
57+
if Task.isCancelled { return }
58+
await mirror(rows)
59+
}
60+
} catch {
61+
Log.error("ProfileMemberMirror: stream failed: \(error.localizedDescription)")
62+
}
63+
}
64+
65+
private func mirror(_ rows: [DBMemberProfile]) async {
66+
guard let selfInboxId = await resolveSelfInboxId() else { return }
67+
let backfill = ProfileBackfill(
68+
databaseReader: databaseReader,
69+
profileStore: profileStore,
70+
selfProfileStore: selfProfileStore,
71+
selfInboxId: selfInboxId
72+
)
73+
do {
74+
try await backfill.mirror(rows)
75+
} catch {
76+
Log.error("ProfileMemberMirror: mirror failed: \(error.localizedDescription)")
77+
}
78+
}
79+
80+
private func resolveSelfInboxId() async -> String? {
81+
if let cachedSelfInboxId { return cachedSelfInboxId }
82+
let resolved = await selfInboxIdProvider()
83+
cachedSelfInboxId = resolved
84+
return resolved
85+
}
86+
}

‎ConvosCore/Sources/ConvosCore/Profiles/Repository/ProfilesRepository.swift‎

Lines changed: 32 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -11,11 +11,12 @@ import Foundation
1111
actor ProfilesRepository {
1212
private let profileStore: any ProfileStoreProtocol
1313
private let selfProfileStore: any SelfProfileStoreProtocol
14-
private let selfInboxId: String
14+
private let selfInboxIdProvider: @Sendable () async -> String?
1515

1616
private var identities: [String: DBProfile] = [:]
1717
private var avatarsByInbox: [String: [String: DBProfileAvatar]] = [:]
1818
private var cachedSelf: DBSelfProfile?
19+
private var cachedSelfInboxId: String?
1920
private var warmedUp: Bool = false
2021

2122
private let changesRelay: ProfileChangesRelay = .init()
@@ -27,18 +28,24 @@ actor ProfilesRepository {
2728
changesRelay.subject.eraseToAnyPublisher()
2829
}
2930

31+
/// `selfInboxIdProvider` resolves the current user's inbox id, which becomes
32+
/// available asynchronously (after inbox-ready). The repository caches the
33+
/// first non-nil result. This lets the DI construct the repository
34+
/// synchronously while the inbox id resolves lazily, matching the
35+
/// `ConnectionServicesStore` pattern.
3036
init(
3137
profileStore: any ProfileStoreProtocol,
3238
selfProfileStore: any SelfProfileStoreProtocol,
33-
selfInboxId: String
39+
selfInboxIdProvider: @escaping @Sendable () async -> String?
3440
) {
3541
self.profileStore = profileStore
3642
self.selfProfileStore = selfProfileStore
37-
self.selfInboxId = selfInboxId
43+
self.selfInboxIdProvider = selfInboxIdProvider
3844
}
3945

4046
func warmUp() async {
4147
guard !warmedUp else { return }
48+
_ = await resolveSelfInboxId()
4249
do {
4350
for identity in try await profileStore.allIdentities() {
4451
identities[identity.inboxId] = identity
@@ -53,6 +60,13 @@ actor ProfilesRepository {
5360
}
5461
}
5562

63+
private func resolveSelfInboxId() async -> String? {
64+
if let cachedSelfInboxId { return cachedSelfInboxId }
65+
let resolved = await selfInboxIdProvider()
66+
cachedSelfInboxId = resolved
67+
return resolved
68+
}
69+
5670
// MARK: - Reads
5771

5872
func profile(inboxId: String) -> UnifiedProfile {
@@ -87,7 +101,12 @@ actor ProfilesRepository {
87101
/// Events authored by the current user are ignored - self identity is held
88102
/// in `selfProfile`, not `DBProfile`.
89103
func apply(_ event: ProfileDomainEvent) async {
90-
guard event.inboxId != selfInboxId else { return }
104+
// Skip the current user's own echoed profile; self identity lives in
105+
// `selfProfile`. If the inbox id isn't resolvable yet, process the event
106+
// rather than risk dropping another member's data.
107+
if let selfId = await resolveSelfInboxId(), event.inboxId == selfId {
108+
return
109+
}
91110
var changed = false
92111

93112
let existingIdentity = identities[event.inboxId]
@@ -135,11 +154,14 @@ actor ProfilesRepository {
135154
// MARK: - Self write
136155

137156
func updateSelfProfile(_ edit: SelfProfileEdit) async throws {
138-
let existing = cachedSelf ?? DBSelfProfile(inboxId: selfInboxId, updatedAt: .distantPast)
157+
guard let selfId = await resolveSelfInboxId() else {
158+
throw ProfilesRepositoryError.selfInboxUnavailable
159+
}
160+
let existing = cachedSelf ?? DBSelfProfile(inboxId: selfId, updatedAt: .distantPast)
139161
let updated = edit.applied(to: existing, updatedAt: Date())
140162
try await selfProfileStore.save(updated)
141163
cachedSelf = updated
142-
changesRelay.subject.send(selfInboxId)
164+
changesRelay.subject.send(selfId)
143165
}
144166

145167
/// Drops a conversation's avatar slots from every person's cache and the
@@ -163,3 +185,7 @@ actor ProfilesRepository {
163185
private final class ProfileChangesRelay: @unchecked Sendable {
164186
let subject: PassthroughSubject<String, Never> = .init()
165187
}
188+
189+
enum ProfilesRepositoryError: Error {
190+
case selfInboxUnavailable
191+
}

‎ConvosCore/Sources/ConvosCore/Sessions/SessionManager.swift‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -171,6 +171,15 @@ public final class SessionManager: SessionManagerProtocol, @unchecked Sendable {
171171
)
172172
await renewalManager.performRenewalIfNeeded()
173173
}
174+
175+
// Populate the canonical Profile tables in the background: a one-time
176+
// backfill from legacy memberProfile, then mirror ongoing writes.
177+
// Self-guards on inbox-ready; the new tables stay dormant (nothing
178+
// renders from them) until the cutover.
179+
Task { [weak self] in
180+
guard let self, !Task.isCancelled else { return }
181+
await self.loadOrCreateService().startProfileMirroring()
182+
}
174183
}
175184
}
176185

@@ -528,6 +537,7 @@ public final class SessionManager: SessionManagerProtocol, @unchecked Sendable {
528537

529538
if let existing {
530539
Log.info("Tearing down authorized inbox")
540+
await existing.stopProfileMirroring()
531541
await existing.stopAndDelete()
532542
await existing.waitForDeletionComplete()
533543
}

‎ConvosCore/Tests/ConvosCoreTests/ProfileBackfillTests.swift‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -106,6 +106,34 @@ struct ProfileBackfillTests {
106106
#expect(alice?.profileSource == .profileUpdate)
107107
}
108108

109+
@Test("mirror(_:) populates the stores from provided rows")
110+
func mirrorFromRows() async throws {
111+
let profileStore = InMemoryProfileStore()
112+
let selfStore = InMemorySelfProfileStore()
113+
let backfill = ProfileBackfill(databaseReader: try DatabaseQueue(), profileStore: profileStore, selfProfileStore: selfStore, selfInboxId: "me")
114+
115+
try await backfill.mirror([
116+
DBMemberProfile(conversationId: "c1", inboxId: "alice", name: "Alice", avatar: "u", avatarSalt: salt, avatarNonce: nonce, avatarKey: key),
117+
])
118+
119+
let alice = try await profileStore.identity(inboxId: "alice")
120+
#expect(alice?.name == "Alice")
121+
let avatar = try await profileStore.avatar(inboxId: "alice", conversationId: "c1")
122+
#expect(avatar?.url == "u")
123+
}
124+
125+
@Test("mirror(_:) tracks a changed value on re-run")
126+
func mirrorTracksChange() async throws {
127+
let profileStore = InMemoryProfileStore()
128+
let backfill = ProfileBackfill(databaseReader: try DatabaseQueue(), profileStore: profileStore, selfProfileStore: InMemorySelfProfileStore(), selfInboxId: "me")
129+
130+
try await backfill.mirror([DBMemberProfile(conversationId: "c1", inboxId: "alice", name: "Alice", avatar: nil)])
131+
try await backfill.mirror([DBMemberProfile(conversationId: "c1", inboxId: "alice", name: "Alicia", avatar: nil)])
132+
133+
let alice = try await profileStore.identity(inboxId: "alice")
134+
#expect(alice?.name == "Alicia")
135+
}
136+
109137
@Test("does nothing when there are no legacy rows")
110138
func emptyIsNoop() async throws {
111139
let queue = try makeMemberProfileQueue()

‎ConvosCore/Tests/ConvosCoreTests/ProfilesRepositoryTests.swift‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,11 @@ struct ProfilesRepositoryTests {
1313
selfProfileStore: any SelfProfileStoreProtocol = InMemorySelfProfileStore(),
1414
selfInboxId: String = "me"
1515
) -> ProfilesRepository {
16-
ProfilesRepository(profileStore: profileStore, selfProfileStore: selfProfileStore, selfInboxId: selfInboxId)
16+
ProfilesRepository(
17+
profileStore: profileStore,
18+
selfProfileStore: selfProfileStore,
19+
selfInboxIdProvider: { selfInboxId }
20+
)
1721
}
1822

1923
private func setAvatar(_ url: String) -> IncomingAvatar {

0 commit comments

Comments
 (0)