Skip to content

Commit 119d8f2

Browse files
cameronvoellcameronvoell
committed
lazy, change-aware profile propagation
1 parent 5a31aad commit 119d8f2

12 files changed

Lines changed: 286 additions & 100 deletions

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

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -183,6 +183,7 @@ final class MessagingService: MessagingServiceProtocol, @unchecked Sendable {
183183
selfProfileStore: selfProfileStore,
184184
publishStore: profilePublishStore,
185185
databaseReader: databaseReader,
186+
conversationLocalStateWriter: ConversationLocalStateWriter(databaseWriter: databaseWriter),
186187
selfInboxIdProvider: { [sessionStateManager] in
187188
(try? await sessionStateManager.waitForInboxReadyResult())?.client.inboxId
188189
}
@@ -314,7 +315,14 @@ final class MessagingService: MessagingServiceProtocol, @unchecked Sendable {
314315
backgroundUploadManager: backgroundUploadManager,
315316
attachmentLocalStateWriter: AttachmentLocalStateWriter(databaseWriter: databaseWriter),
316317
contactSyncCoordinator: contactSyncCoordinator(),
317-
coreActions: coreActions
318+
coreActions: coreActions,
319+
ensureProfilePublished: { [sharedProfilesRepository] in
320+
do {
321+
try await sharedProfilesRepository.publishMyProfileToConversation(conversationId)
322+
} catch {
323+
Log.warning("Failed to publish profile before send to \(conversationId): \(error.localizedDescription)")
324+
}
325+
}
318326
)
319327
}
320328

‎ConvosCore/Sources/ConvosCore/Mocks/MockConversationLocalStateWriter.swift‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ public final class MockConversationLocalStateWriter: ConversationLocalStateWrite
88
public var hidesInviteCardStates: [String: Bool] = [:]
99
public var leftHostedInviteSessionStates: [String: Bool] = [:]
1010
public var hasSharedInviteStates: [String: Bool] = [:]
11+
public var publishedProfileUpdatedAtStates: [String: Date?] = [:]
1112

1213
public init() {}
1314

@@ -34,4 +35,8 @@ public final class MockConversationLocalStateWriter: ConversationLocalStateWrite
3435
public func setHasSharedInvite(_ hasSharedInvite: Bool, for conversationId: String) async throws {
3536
hasSharedInviteStates[conversationId] = hasSharedInvite
3637
}
38+
39+
public func setPublishedProfileUpdatedAt(_ publishedProfileUpdatedAt: Date?, for conversationId: String) async throws {
40+
publishedProfileUpdatedAtStates[conversationId] = publishedProfileUpdatedAt
41+
}
3742
}

‎ConvosCore/Sources/ConvosCore/Mocks/MockMessagingService.swift‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,7 @@ public final class MockMessagingService: MessagingServiceProtocol, @unchecked Se
6969
selfProfileStore: InMemorySelfProfileStore(),
7070
publishStore: InMemoryProfilePublishStore(),
7171
databaseReader: MockDatabaseManager.previews.dbReader,
72+
conversationLocalStateWriter: ConversationLocalStateWriter(databaseWriter: MockDatabaseManager.previews.dbWriter),
7273
selfInboxIdProvider: { nil }
7374
)
7475
}

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

Lines changed: 9 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -70,28 +70,16 @@ actor ProfilePublisher {
7070
session = nil
7171
}
7272

73-
/// Records a new source avatar (when `avatarBytes` is provided) and enqueues
74-
/// a publish job per conversation, priority conversation first, then drains.
75-
/// A nil `avatarBytes` is a name/metadata-only publish that re-sends the
76-
/// existing avatar for each conversation rather than clearing it.
77-
func publish(avatarBytes: Data?, priorityConversationId: String?) async throws {
73+
/// Records a new source avatar so subsequent per-conversation publishes
74+
/// re-encrypt it to each conversation's group key. Does not enqueue or fan
75+
/// out - propagation is lazy and per-conversation via `publishConversation`.
76+
func updateAvatarSource(_ avatarBytes: Data) async throws {
7877
guard let selfInboxId = await resolveSelfInboxId() else { return }
79-
let conversationIds = try await session?.conversationIds() ?? []
80-
var sourceVersion: Int64?
81-
var hasAvatar = false
82-
if let avatarBytes {
83-
let existing = try await publishStore.source(inboxId: selfInboxId)
84-
let version = (existing?.version ?? 0) + 1
85-
try await publishStore.setSource(
86-
DBProfileAvatarSource(inboxId: selfInboxId, plaintext: avatarBytes, version: version, updatedAt: now())
87-
)
88-
sourceVersion = version
89-
hasAvatar = true
90-
}
91-
for conversationId in ordered(conversationIds, priority: priorityConversationId) {
92-
try await enqueueJob(conversationId: conversationId, sourceVersion: sourceVersion, hasAvatar: hasAvatar)
93-
}
94-
await drainReadyJobs()
78+
let existing = try await publishStore.source(inboxId: selfInboxId)
79+
let version = (existing?.version ?? 0) + 1
80+
try await publishStore.setSource(
81+
DBProfileAvatarSource(inboxId: selfInboxId, plaintext: avatarBytes, version: version, updatedAt: now())
82+
)
9583
}
9684

9785
/// Seeds a single conversation with the current profile (e.g. a freshly
@@ -133,11 +121,6 @@ actor ProfilePublisher {
133121
try await publishStore.supersedeOlderThan(conversationId: conversationId, seq: seq)
134122
}
135123

136-
private func ordered(_ ids: [String], priority: String?) -> [String] {
137-
guard let priority, ids.contains(priority) else { return ids }
138-
return [priority] + ids.filter { $0 != priority }
139-
}
140-
141124
// MARK: - Process
142125

143126
private func process(_ job: DBProfilePublishJob, session: any ProfilePublishSession, selfInboxId: String) async {

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

Lines changed: 54 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ public actor ProfilesRepository {
1414
private let profileStore: any ProfileStoreProtocol
1515
private let selfProfileStore: any SelfProfileStoreProtocol
1616
private let databaseReader: any DatabaseReader
17+
private let conversationLocalStateWriter: any ConversationLocalStateWriterProtocol
1718
private let selfInboxIdProvider: @Sendable () async -> String?
1819
private let publisher: ProfilePublisher
1920

@@ -42,11 +43,13 @@ public actor ProfilesRepository {
4243
selfProfileStore: any SelfProfileStoreProtocol,
4344
publishStore: any ProfilePublishStoreProtocol,
4445
databaseReader: any DatabaseReader,
46+
conversationLocalStateWriter: any ConversationLocalStateWriterProtocol,
4547
selfInboxIdProvider: @escaping @Sendable () async -> String?
4648
) {
4749
self.profileStore = profileStore
4850
self.selfProfileStore = selfProfileStore
4951
self.databaseReader = databaseReader
52+
self.conversationLocalStateWriter = conversationLocalStateWriter
5053
self.selfInboxIdProvider = selfInboxIdProvider
5154
self.publisher = ProfilePublisher(
5255
publishStore: publishStore,
@@ -256,38 +259,69 @@ public actor ProfilesRepository {
256259
await publisher.detach()
257260
}
258261

259-
/// Records a name and/or new avatar and fans the update out to every
260-
/// conversation via the durable publish queue. The name is written to the
261-
/// self profile first so the publisher's jobs send the current name.
262+
/// Records a name and/or new avatar edit locally and bumps the self
263+
/// profile's `updatedAt`. Propagation is lazy: the edit reaches a
264+
/// conversation only when the user next opens or sends in it (via
265+
/// `publishMyProfileToConversation`), so it never fans out to conversations
266+
/// the user has stopped engaging with. When a `priorityConversationId` (the
267+
/// active conversation) is given, it is published immediately so the user
268+
/// sees their change reflected without waiting for the next send.
262269
public func publishMyProfile(displayName: String?, avatarBytes: Data?, priorityConversationId: String?) async throws {
270+
let nameField: SelfProfileEdit.Field<String?>
263271
if let displayName {
264-
try await updateSelfProfile(SelfProfileEdit(name: .set(displayName)))
272+
nameField = .set(displayName)
273+
} else {
274+
nameField = .keep
275+
}
276+
try await updateSelfProfile(SelfProfileEdit(name: nameField))
277+
if let avatarBytes {
278+
try await publisher.updateAvatarSource(avatarBytes)
279+
}
280+
if let priorityConversationId {
281+
try await publishMyProfileToConversation(priorityConversationId)
265282
}
266-
try await publisher.publish(avatarBytes: avatarBytes, priorityConversationId: priorityConversationId)
267283
}
268284

269-
/// Seeds a freshly created or joined conversation with the current profile.
270-
///
271-
/// Idempotent: skips when the current user already has an avatar slot for
272-
/// this conversation, i.e. we have already published our profile here. This
273-
/// prevents a redundant `ProfileUpdate` on every conversation open. Skipping
274-
/// cannot drop a real change - a genuine profile edit fans out to every known
275-
/// conversation via `publishMyProfile`, not the seeder. (A name-only user has
276-
/// no avatar slot, so they may re-seed on open; a cheap, known residual.)
285+
/// Publishes the current self profile to one conversation, but only if it has
286+
/// changed since the last profile published there. The comparison is
287+
/// `ConversationLocalState.publishedProfileUpdatedAt` (what we last sent here)
288+
/// against `DBMyProfile.updatedAt` (the current profile). This is the lazy
289+
/// propagation seam: a fresh conversation (no stamp) always publishes, an
290+
/// up-to-date conversation is a no-op, and a stale one re-publishes and
291+
/// re-stamps. Called on conversation open and before an outgoing send, so it
292+
/// is impossible to send a message after editing your profile without also
293+
/// sending the profile update.
277294
public func publishMyProfileToConversation(_ conversationId: String) async throws {
278-
if let selfId = await resolveSelfInboxId(),
279-
let existing = try? await profileStore.avatar(inboxId: selfId, conversationId: conversationId),
280-
existing.url != nil {
281-
return
295+
guard let selfId = await resolveSelfInboxId() else { return }
296+
let decision: (shouldPublish: Bool, target: Date?) = try await databaseReader.read { db -> (shouldPublish: Bool, target: Date?) in
297+
guard let myUpdatedAt = try DBMyProfile
298+
.filter(DBMyProfile.Columns.inboxId == selfId)
299+
.fetchOne(db)?.updatedAt else {
300+
return (false, nil)
301+
}
302+
let publishedAt = try ConversationLocalState
303+
.filter(ConversationLocalState.Columns.conversationId == conversationId)
304+
.fetchOne(db)?.publishedProfileUpdatedAt
305+
let stale: Bool
306+
if let publishedAt {
307+
stale = publishedAt < myUpdatedAt
308+
} else {
309+
stale = true
310+
}
311+
return (stale, myUpdatedAt)
282312
}
313+
guard decision.shouldPublish, let target = decision.target else { return }
283314
try await publisher.publishConversation(conversationId)
315+
// Optimistic stamp: the publish job is durable and retries until it
316+
// lands, so recording it now prevents re-enqueuing on the next
317+
// open/send while the profile is unchanged.
318+
try await conversationLocalStateWriter.setPublishedProfileUpdatedAt(target, for: conversationId)
284319
}
285320

286-
/// Records new self metadata and re-publishes it (a name/metadata-only fan-out
287-
/// that re-sends the existing avatar) to every conversation.
321+
/// Records new self metadata locally and bumps `updatedAt`. Propagation is
322+
/// lazy per conversation, same as `publishMyProfile`.
288323
public func publishMyProfileMetadata(_ metadata: ProfileMetadata?) async throws {
289324
try await updateSelfProfile(SelfProfileEdit(metadata: .set(metadata)))
290-
try await publisher.publish(avatarBytes: nil, priorityConversationId: nil)
291325
}
292326

293327
/// Drops a conversation's avatar slots from every person's cache and the

‎ConvosCore/Sources/ConvosCore/Storage/Database Models/ConversationLocalState.swift‎

Lines changed: 70 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ struct ConversationLocalState: Codable, FetchableRecord, PersistableRecord, Hash
1818
static let wasRemoved: Column = Column(CodingKeys.wasRemoved)
1919
static let hasHadOtherMembers: Column = Column(CodingKeys.hasHadOtherMembers)
2020
static let hasSharedInvite: Column = Column(CodingKeys.hasSharedInvite)
21+
static let publishedProfileUpdatedAt: Column = Column(CodingKeys.publishedProfileUpdatedAt)
2122
}
2223

2324
let conversationId: String
@@ -46,6 +47,41 @@ struct ConversationLocalState: Codable, FetchableRecord, PersistableRecord, Hash
4647
/// Local-only, so network sync can never clobber it; deleted with the
4748
/// conversation. Read by `ConversationEngagement.isEngaged`.
4849
let hasSharedInvite: Bool
50+
/// The `DBMyProfile.updatedAt` of the self profile last published to this
51+
/// conversation. The lazy profile sync compares this against the current
52+
/// `DBMyProfile.updatedAt` to decide whether an edit still needs to be sent
53+
/// here, so a profile change only reaches conversations the user re-engages
54+
/// rather than fanning out to every conversation. nil means never published
55+
/// (treated as stale). Local-only; deleted with the conversation.
56+
let publishedProfileUpdatedAt: Date?
57+
58+
init(
59+
conversationId: String,
60+
isPinned: Bool,
61+
isUnread: Bool,
62+
isUnreadUpdatedAt: Date,
63+
isMuted: Bool,
64+
pinnedOrder: Int?,
65+
hidesInviteCard: Bool,
66+
leftHostedInviteSession: Bool,
67+
wasRemoved: Bool,
68+
hasHadOtherMembers: Bool,
69+
hasSharedInvite: Bool,
70+
publishedProfileUpdatedAt: Date? = nil
71+
) {
72+
self.conversationId = conversationId
73+
self.isPinned = isPinned
74+
self.isUnread = isUnread
75+
self.isUnreadUpdatedAt = isUnreadUpdatedAt
76+
self.isMuted = isMuted
77+
self.pinnedOrder = pinnedOrder
78+
self.hidesInviteCard = hidesInviteCard
79+
self.leftHostedInviteSession = leftHostedInviteSession
80+
self.wasRemoved = wasRemoved
81+
self.hasHadOtherMembers = hasHadOtherMembers
82+
self.hasSharedInvite = hasSharedInvite
83+
self.publishedProfileUpdatedAt = publishedProfileUpdatedAt
84+
}
4985

5086
static let conversationForeignKey: ForeignKey = ForeignKey([Columns.conversationId], to: [DBConversation.Columns.id])
5187

@@ -68,7 +104,8 @@ extension ConversationLocalState {
68104
leftHostedInviteSession: leftHostedInviteSession,
69105
wasRemoved: wasRemoved,
70106
hasHadOtherMembers: hasHadOtherMembers,
71-
hasSharedInvite: hasSharedInvite
107+
hasSharedInvite: hasSharedInvite,
108+
publishedProfileUpdatedAt: publishedProfileUpdatedAt
72109
)
73110
}
74111
func with(isPinned: Bool) -> Self {
@@ -83,7 +120,8 @@ extension ConversationLocalState {
83120
leftHostedInviteSession: leftHostedInviteSession,
84121
wasRemoved: wasRemoved,
85122
hasHadOtherMembers: hasHadOtherMembers,
86-
hasSharedInvite: hasSharedInvite
123+
hasSharedInvite: hasSharedInvite,
124+
publishedProfileUpdatedAt: publishedProfileUpdatedAt
87125
)
88126
}
89127
func with(isMuted: Bool) -> Self {
@@ -98,7 +136,8 @@ extension ConversationLocalState {
98136
leftHostedInviteSession: leftHostedInviteSession,
99137
wasRemoved: wasRemoved,
100138
hasHadOtherMembers: hasHadOtherMembers,
101-
hasSharedInvite: hasSharedInvite
139+
hasSharedInvite: hasSharedInvite,
140+
publishedProfileUpdatedAt: publishedProfileUpdatedAt
102141
)
103142
}
104143
func with(pinnedOrder: Int?) -> Self {
@@ -113,7 +152,8 @@ extension ConversationLocalState {
113152
leftHostedInviteSession: leftHostedInviteSession,
114153
wasRemoved: wasRemoved,
115154
hasHadOtherMembers: hasHadOtherMembers,
116-
hasSharedInvite: hasSharedInvite
155+
hasSharedInvite: hasSharedInvite,
156+
publishedProfileUpdatedAt: publishedProfileUpdatedAt
117157
)
118158
}
119159
func with(hidesInviteCard: Bool) -> Self {
@@ -128,7 +168,8 @@ extension ConversationLocalState {
128168
leftHostedInviteSession: leftHostedInviteSession,
129169
wasRemoved: wasRemoved,
130170
hasHadOtherMembers: hasHadOtherMembers,
131-
hasSharedInvite: hasSharedInvite
171+
hasSharedInvite: hasSharedInvite,
172+
publishedProfileUpdatedAt: publishedProfileUpdatedAt
132173
)
133174
}
134175
func with(leftHostedInviteSession: Bool) -> Self {
@@ -143,7 +184,8 @@ extension ConversationLocalState {
143184
leftHostedInviteSession: leftHostedInviteSession,
144185
wasRemoved: wasRemoved,
145186
hasHadOtherMembers: hasHadOtherMembers,
146-
hasSharedInvite: hasSharedInvite
187+
hasSharedInvite: hasSharedInvite,
188+
publishedProfileUpdatedAt: publishedProfileUpdatedAt
147189
)
148190
}
149191
func with(wasRemoved: Bool) -> Self {
@@ -158,7 +200,8 @@ extension ConversationLocalState {
158200
leftHostedInviteSession: leftHostedInviteSession,
159201
wasRemoved: wasRemoved,
160202
hasHadOtherMembers: hasHadOtherMembers,
161-
hasSharedInvite: hasSharedInvite
203+
hasSharedInvite: hasSharedInvite,
204+
publishedProfileUpdatedAt: publishedProfileUpdatedAt
162205
)
163206
}
164207
func with(hasHadOtherMembers: Bool) -> Self {
@@ -173,7 +216,8 @@ extension ConversationLocalState {
173216
leftHostedInviteSession: leftHostedInviteSession,
174217
wasRemoved: wasRemoved,
175218
hasHadOtherMembers: hasHadOtherMembers,
176-
hasSharedInvite: hasSharedInvite
219+
hasSharedInvite: hasSharedInvite,
220+
publishedProfileUpdatedAt: publishedProfileUpdatedAt
177221
)
178222
}
179223
func with(hasSharedInvite: Bool) -> Self {
@@ -188,7 +232,24 @@ extension ConversationLocalState {
188232
leftHostedInviteSession: leftHostedInviteSession,
189233
wasRemoved: wasRemoved,
190234
hasHadOtherMembers: hasHadOtherMembers,
191-
hasSharedInvite: hasSharedInvite
235+
hasSharedInvite: hasSharedInvite,
236+
publishedProfileUpdatedAt: publishedProfileUpdatedAt
237+
)
238+
}
239+
func with(publishedProfileUpdatedAt: Date?) -> Self {
240+
.init(
241+
conversationId: conversationId,
242+
isPinned: isPinned,
243+
isUnread: isUnread,
244+
isUnreadUpdatedAt: isUnreadUpdatedAt,
245+
isMuted: isMuted,
246+
pinnedOrder: pinnedOrder,
247+
hidesInviteCard: hidesInviteCard,
248+
leftHostedInviteSession: leftHostedInviteSession,
249+
wasRemoved: wasRemoved,
250+
hasHadOtherMembers: hasHadOtherMembers,
251+
hasSharedInvite: hasSharedInvite,
252+
publishedProfileUpdatedAt: publishedProfileUpdatedAt
192253
)
193254
}
194255
}

‎ConvosCore/Sources/ConvosCore/Storage/SharedDatabaseMigrator.swift‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -264,6 +264,22 @@ extension SharedDatabaseMigrator {
264264
migrator.registerMigration("dropSelfProfileTable", migrate: Self.dropSelfProfileTable)
265265
migrator.registerMigration("addProfileAvatarLastRenewed", migrate: Self.addProfileAvatarLastRenewed)
266266
migrator.registerMigration("makeProfileAvatarLatestViewDeterministic", migrate: Self.makeProfileAvatarLatestViewDeterministic)
267+
migrator.registerMigration("addConversationLocalStatePublishedProfileUpdatedAt", migrate: Self.addConversationLocalStatePublishedProfileUpdatedAt)
268+
}
269+
270+
/// Additive, nullable column recording the `myProfile.updatedAt` of the self
271+
/// profile last published to a conversation. The lazy profile sync compares
272+
/// it against the current `myProfile.updatedAt` to decide whether a profile
273+
/// edit still needs to be sent to this conversation, so edits reach only
274+
/// conversations the user re-engages rather than fanning out to all of them.
275+
/// `nil` means never published (treated as stale), so no backfill is needed.
276+
/// Guarded on column existence for idempotency across dev installs.
277+
static func addConversationLocalStatePublishedProfileUpdatedAt(_ db: Database) throws {
278+
let hasColumn = try db.columns(in: "conversationLocalState").contains { $0.name == "publishedProfileUpdatedAt" }
279+
guard !hasColumn else { return }
280+
try db.alter(table: "conversationLocalState") { t in
281+
t.add(column: "publishedProfileUpdatedAt", .datetime)
282+
}
267283
}
268284

269285
/// Recreates `profileAvatarLatest` with the deterministic `ROW_NUMBER()`

0 commit comments

Comments
 (0)