Skip to content

Commit 4fa3f85

Browse files
cameronvoellcameronvoell
committed
unified profile: wire repository reads, self-publish, and self-profile freshness
1 parent 69b95b5 commit 4fa3f85

10 files changed

Lines changed: 530 additions & 40 deletions

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

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -174,9 +174,16 @@ final class MessagingService: MessagingServiceProtocol, @unchecked Sendable {
174174
databaseReader: databaseReader
175175
)
176176

177+
private lazy var profilePublishStore: any ProfilePublishStoreProtocol = GRDBProfilePublishStore(
178+
databaseWriter: databaseWriter,
179+
databaseReader: databaseReader
180+
)
181+
177182
private lazy var sharedProfilesRepository: ProfilesRepository = ProfilesRepository(
178183
profileStore: profileStore,
179184
selfProfileStore: selfProfileStore,
185+
publishStore: profilePublishStore,
186+
databaseReader: databaseReader,
180187
selfInboxIdProvider: { [sessionStateManager] in
181188
(try? await sessionStateManager.waitForInboxReadyResult())?.client.inboxId
182189
}
@@ -191,6 +198,11 @@ final class MessagingService: MessagingServiceProtocol, @unchecked Sendable {
191198
}
192199
)
193200

201+
private lazy var profileShadowComparator: ProfileShadowComparator = ProfileShadowComparator(
202+
databaseReader: databaseReader,
203+
profileStore: profileStore
204+
)
205+
194206
/// Canonical identity source. Dormant until the cutover - nothing renders
195207
/// from it yet.
196208
func profilesRepository() -> ProfilesRepository {
@@ -215,11 +227,20 @@ final class MessagingService: MessagingServiceProtocol, @unchecked Sendable {
215227
}
216228
}
217229
await sharedProfilesRepository.warmUp()
230+
await sharedProfilesRepository.bind(
231+
session: MessagingProfilePublishSession(
232+
sessionStateManager: sessionStateManager,
233+
databaseReader: databaseReader
234+
)
235+
)
218236
await profileMemberMirror.start()
237+
await profileShadowComparator.start()
219238
}
220239

221240
func stopProfileMirroring() async {
241+
await sharedProfilesRepository.unbind()
222242
await profileMemberMirror.stop()
243+
await profileShadowComparator.stop()
223244
}
224245

225246
// MARK: New Conversation

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

Lines changed: 24 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,11 @@ import GRDB
1111
/// and non-clobbering: re-running fills only blanks of whatever is already there,
1212
/// and a value already set by a real `profileUpdate` is never downgraded.
1313
///
14-
/// Not wired into startup yet; the lifecycle PR calls `run()`.
14+
/// The current user's own row is the exception: the self profile has no source
15+
/// column and, until the cutover, the legacy `memberProfile` is its sole author,
16+
/// so the self row is upserted (not seed-once) to reflect renames on re-run.
17+
///
18+
/// Also re-run on every `memberProfile` change via `ProfileMemberMirror`.
1519
struct ProfileBackfill {
1620
private let databaseReader: any DatabaseReader
1721
private let profileStore: any ProfileStoreProtocol
@@ -78,12 +82,29 @@ struct ProfileBackfill {
7882
}
7983

8084
try await backfillIdentities(identityByInbox)
85+
try await mirrorSelf(sawSelf: sawSelf, name: selfName, metadata: selfMetadata)
86+
}
8187

82-
if sawSelf, try await selfProfileStore.load() == nil {
88+
/// Upserts the self row from the legacy-derived name/metadata. A fresh seed
89+
/// uses the floor timestamp (like everything else here); an update reflects a
90+
/// legacy rename with a real timestamp. Legacy values win when present but a
91+
/// blank never erases an existing value, and the write is skipped when nothing
92+
/// changed so re-runs don't churn.
93+
private func mirrorSelf(sawSelf: Bool, name: String?, metadata: ProfileMetadata?) async throws {
94+
guard sawSelf else { return }
95+
let existing = try await selfProfileStore.load()
96+
let mergedName = ProfileMerge.nonBlank(name) ?? existing?.name
97+
let mergedMetadata = metadata ?? existing?.metadata
98+
guard let existing else {
8399
try await selfProfileStore.save(
84-
DBSelfProfile(inboxId: selfInboxId, name: selfName, metadata: selfMetadata, updatedAt: floor)
100+
DBSelfProfile(inboxId: selfInboxId, name: mergedName, metadata: mergedMetadata, updatedAt: floor)
85101
)
102+
return
86103
}
104+
guard mergedName != existing.name || mergedMetadata != existing.metadata else { return }
105+
try await selfProfileStore.save(
106+
DBSelfProfile(inboxId: selfInboxId, name: mergedName, metadata: mergedMetadata, updatedAt: Date())
107+
)
87108
}
88109

89110
private func backfillAvatar(_ row: DBMemberProfile) async throws {

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

Lines changed: 23 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -23,35 +23,44 @@ struct PublishBackoff: Sendable {
2323
/// loop is never head-of-line blocked. XMTP and upload specifics live behind the
2424
/// injected `ProfilePublishSession`.
2525
///
26-
/// Not wired into startup yet; the lifecycle PR attaches a session and triggers
27-
/// drains on a wake timer. Within a process, drains are serialized.
26+
/// `ProfilesRepository` owns the instance and attaches a session via
27+
/// `bind(session:)` at startup, which resumes any leftover jobs. Within a
28+
/// process, drains are serialized.
2829
actor ProfilePublisher {
2930
private let publishStore: any ProfilePublishStoreProtocol
3031
private let profileStore: any ProfileStoreProtocol
3132
private let selfProfileStore: any SelfProfileStoreProtocol
32-
private let selfInboxId: String
33+
private let selfInboxIdProvider: @Sendable () async -> String?
3334
private let now: @Sendable () -> Date
3435
private let backoff: PublishBackoff
3536

3637
private var session: (any ProfilePublishSession)?
3738
private var draining: Bool = false
39+
private var cachedSelfInboxId: String?
3840

3941
init(
4042
publishStore: any ProfilePublishStoreProtocol,
4143
profileStore: any ProfileStoreProtocol,
4244
selfProfileStore: any SelfProfileStoreProtocol,
43-
selfInboxId: String,
45+
selfInboxIdProvider: @escaping @Sendable () async -> String?,
4446
now: @escaping @Sendable () -> Date = { Date() },
4547
backoff: PublishBackoff = .default
4648
) {
4749
self.publishStore = publishStore
4850
self.profileStore = profileStore
4951
self.selfProfileStore = selfProfileStore
50-
self.selfInboxId = selfInboxId
52+
self.selfInboxIdProvider = selfInboxIdProvider
5153
self.now = now
5254
self.backoff = backoff
5355
}
5456

57+
private func resolveSelfInboxId() async -> String? {
58+
if let cachedSelfInboxId { return cachedSelfInboxId }
59+
let resolved = await selfInboxIdProvider()
60+
cachedSelfInboxId = resolved
61+
return resolved
62+
}
63+
5564
func attach(session: any ProfilePublishSession) async {
5665
self.session = session
5766
await drainReadyJobs()
@@ -66,6 +75,7 @@ actor ProfilePublisher {
6675
/// A nil `avatarBytes` is a name/metadata-only publish that re-sends the
6776
/// existing avatar for each conversation rather than clearing it.
6877
func publish(avatarBytes: Data?, priorityConversationId: String?) async throws {
78+
guard let selfInboxId = await resolveSelfInboxId() else { return }
6979
let conversationIds = try await session?.conversationIds() ?? []
7080
var sourceVersion: Int64?
7181
var hasAvatar = false
@@ -87,6 +97,7 @@ actor ProfilePublisher {
8797
/// Seeds a single conversation with the current profile (e.g. a freshly
8898
/// created or joined group), then drains.
8999
func publishConversation(_ conversationId: String) async throws {
100+
guard let selfInboxId = await resolveSelfInboxId() else { return }
90101
let source = try await publishStore.source(inboxId: selfInboxId)
91102
try await enqueueJob(conversationId: conversationId, sourceVersion: source?.version, hasAvatar: source != nil)
92103
await drainReadyJobs()
@@ -96,8 +107,9 @@ actor ProfilePublisher {
96107
guard !draining, let session else { return }
97108
draining = true
98109
defer { draining = false }
110+
guard let selfInboxId = await resolveSelfInboxId() else { return }
99111
while let job = try? await publishStore.nextReadyJob(now: now()) {
100-
await process(job, session: session)
112+
await process(job, session: session, selfInboxId: selfInboxId)
101113
}
102114
}
103115

@@ -128,12 +140,12 @@ actor ProfilePublisher {
128140

129141
// MARK: - Process
130142

131-
private func process(_ job: DBProfilePublishJob, session: any ProfilePublishSession) async {
143+
private func process(_ job: DBProfilePublishJob, session: any ProfilePublishSession, selfInboxId: String) async {
132144
do {
133145
if job.hasAvatar {
134-
try await processAvatarJob(job, session: session)
146+
try await processAvatarJob(job, session: session, selfInboxId: selfInboxId)
135147
} else {
136-
try await processNameOnlyJob(job, session: session)
148+
try await processNameOnlyJob(job, session: session, selfInboxId: selfInboxId)
137149
}
138150
try? await publishStore.deleteJob(id: job.id)
139151
} catch is PublishDropError {
@@ -143,7 +155,7 @@ actor ProfilePublisher {
143155
}
144156
}
145157

146-
private func processAvatarJob(_ job: DBProfilePublishJob, session: any ProfilePublishSession) async throws {
158+
private func processAvatarJob(_ job: DBProfilePublishJob, session: any ProfilePublishSession, selfInboxId: String) async throws {
147159
guard let version = job.sourceVersion,
148160
let source = try await publishStore.source(inboxId: selfInboxId),
149161
source.version == version else {
@@ -198,7 +210,7 @@ actor ProfilePublisher {
198210
try await profileStore.saveAvatar(slot)
199211
}
200212

201-
private func processNameOnlyJob(_ job: DBProfilePublishJob, session: any ProfilePublishSession) async throws {
213+
private func processNameOnlyJob(_ job: DBProfilePublishJob, session: any ProfilePublishSession, selfInboxId: String) async throws {
202214
let existing = try await profileStore.avatar(inboxId: selfInboxId, conversationId: job.conversationId)
203215
let published = publishedAvatar(from: existing)
204216
let selfName = try await selfProfileStore.load()?.name

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

Lines changed: 129 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,129 @@
1+
import Foundation
2+
import GRDB
3+
4+
/// Result of one shadow comparison between the new canonical identity
5+
/// (`DBProfile` / `DBProfileAvatar`) and the legacy `contact` table, over the
6+
/// inboxes present in both.
7+
struct ProfileShadowComparison: Sendable {
8+
let comparedCount: Int
9+
let nameMismatches: Int
10+
let avatarMismatches: Int
11+
let sampleInboxIds: [String]
12+
13+
var hasDiscrepancies: Bool {
14+
nameMismatches > 0 || avatarMismatches > 0
15+
}
16+
17+
var summary: String {
18+
"compared \(comparedCount), name mismatches \(nameMismatches), avatar mismatches \(avatarMismatches), samples \(sampleInboxIds)"
19+
}
20+
}
21+
22+
/// Runs a background "shadow" comparison of the new profile identity against the
23+
/// legacy `contact` table and logs discrepancies. The point is to surface
24+
/// mirror/backfill bugs while the new system is still dormant, so they can be
25+
/// fixed before the cutover flips reads to the repository. Read-only; changes
26+
/// nothing.
27+
///
28+
/// Removed once the cutover is proven and the legacy tables are gone.
29+
actor ProfileShadowComparator {
30+
private let databaseReader: any DatabaseReader
31+
private let profileStore: any ProfileStoreProtocol
32+
private let interval: TimeInterval
33+
private var task: Task<Void, Never>?
34+
35+
init(
36+
databaseReader: any DatabaseReader,
37+
profileStore: any ProfileStoreProtocol,
38+
interval: TimeInterval = 1_800
39+
) {
40+
self.databaseReader = databaseReader
41+
self.profileStore = profileStore
42+
self.interval = interval
43+
}
44+
45+
func start() {
46+
guard task == nil else { return }
47+
let sleepNanos = UInt64(interval * 1_000_000_000)
48+
task = Task { [weak self] in
49+
while !Task.isCancelled {
50+
await self?.compareAndLog()
51+
do {
52+
try await Task.sleep(nanoseconds: sleepNanos)
53+
} catch {
54+
return
55+
}
56+
}
57+
}
58+
}
59+
60+
func stop() {
61+
task?.cancel()
62+
task = nil
63+
}
64+
65+
private func compareAndLog() async {
66+
do {
67+
let result = try await compare()
68+
if result.hasDiscrepancies {
69+
Log.warning("ProfileShadowCompare: \(result.summary)")
70+
} else {
71+
Log.info("ProfileShadowCompare: no discrepancies across \(result.comparedCount) inboxes")
72+
}
73+
} catch {
74+
Log.error("ProfileShadowCompare failed: \(error.localizedDescription)")
75+
}
76+
}
77+
78+
/// Compares name and avatar-presence for every inbox present in both the new
79+
/// stores and the legacy `contact` table.
80+
func compare() async throws -> ProfileShadowComparison {
81+
let profiles = try await profileStore.allIdentities()
82+
let avatars = try await profileStore.allAvatars()
83+
let newAvatarInboxIds = Set(avatars.compactMap { $0.url == nil ? nil : $0.inboxId })
84+
85+
let contacts = try await databaseReader.read { db in
86+
try Row.fetchAll(db, sql: "SELECT inboxId, displayName, avatarURL FROM contact")
87+
.map { row in
88+
ContactRow(inboxId: row["inboxId"], displayName: row["displayName"], avatarURL: row["avatarURL"])
89+
}
90+
}
91+
let contactByInbox = Dictionary(contacts.map { ($0.inboxId, $0) }, uniquingKeysWith: { first, _ in first })
92+
93+
var comparedCount = 0
94+
var nameMismatches = 0
95+
var avatarMismatches = 0
96+
var sampleInboxIds: [String] = []
97+
for profile in profiles {
98+
guard let contact = contactByInbox[profile.inboxId] else { continue }
99+
comparedCount += 1
100+
let nameDiffers = Self.normalized(profile.name) != Self.normalized(contact.displayName)
101+
let avatarDiffers = newAvatarInboxIds.contains(profile.inboxId) != (contact.avatarURL != nil)
102+
if nameDiffers {
103+
nameMismatches += 1
104+
}
105+
if avatarDiffers {
106+
avatarMismatches += 1
107+
}
108+
if nameDiffers || avatarDiffers, sampleInboxIds.count < 5 {
109+
sampleInboxIds.append(profile.inboxId)
110+
}
111+
}
112+
return ProfileShadowComparison(
113+
comparedCount: comparedCount,
114+
nameMismatches: nameMismatches,
115+
avatarMismatches: avatarMismatches,
116+
sampleInboxIds: sampleInboxIds
117+
)
118+
}
119+
120+
private static func normalized(_ value: String?) -> String {
121+
value?.trimmingCharacters(in: .whitespacesAndNewlines) ?? ""
122+
}
123+
}
124+
125+
private struct ContactRow {
126+
let inboxId: String
127+
let displayName: String?
128+
let avatarURL: String?
129+
}

0 commit comments

Comments
 (0)