diff --git a/docs/agent-runbook.md b/docs/agent-runbook.md index 9f2c604b..13d1eb30 100644 --- a/docs/agent-runbook.md +++ b/docs/agent-runbook.md @@ -146,6 +146,27 @@ are re-linked by peer when their thread shows up under its new id. Watch in `/api/status` — a burst right after a re-pair is the fix working, not a fault. +A roster match is not proof of a re-key: two groups with the same members, or +two 1:1 threads with one person, are both live under their own ids. Until +2026-10-08 ingest moved one's id onto the other, so the id and title flipped on +every event and both threads' messages piled into one row. A thread now takes a +new id only when its own id is dead: displaced by a collision, or announced by +a ConversationEvent only in an earlier device ID space. ID spaces are counted +per account (`accounts.remote_idspace_epoch`); a new one starts when an id the +phone announced in the current space arrives naming a different thread. Group +re-keys also need the title to match. A thread whose first messages arrive +with no ConversationEvent is provisional; a second sender makes it a group, so +it is not scattered across its members' 1:1 threads. A reset shows up as the +first collision, and fresh ids that arrive before it (14 threads in the 36 +seconds before the first collision on 9/3) first get their own rows. When the +reset is detected, rows minted that way in the previous 15 minutes are merged +into the threads they continue. Google also reuses message ids across a reset; when a +continued thread receives a new message under an old message's id, the old +message keeps its row under `displaced::`. Counters: +`idspace_resets` (expect one per phone swap or restore), `rekeys_declined` +(a thread kept apart from a live same-roster twin; steady trickle), +`rekeys_recovered`, `remote_message_ids_retired`. + **Repairing history after a reset** (daemon must be down — the command takes the instance lock and refuses while `/api/status` answers; park the watchdog first): diff --git a/internal/ingest/counters.go b/internal/ingest/counters.go index 7df6e25a..22aa3237 100644 --- a/internal/ingest/counters.go +++ b/internal/ingest/counters.go @@ -38,6 +38,23 @@ type CounterSnapshot struct { // dropped because identical content already existed under another remote id. RemoteRebinds uint64 `json:"remote_rebinds"` ContentDupesSkipped uint64 `json:"content_dupes_skipped"` + + // IDSpaceResets counts Google device ID-space resets detected: a wire id + // the phone had announced for one thread in the current ID space arrived + // naming a different one. RekeysDeclined counts frames whose roster matched + // only threads still live under their own ids (two groups with one + // membership, two 1:1 threads with one person), so the frame's id got its + // own thread instead of taking theirs. RekeysRecovered counts threads kept + // apart that way in a reset's opening minutes, before the reset was + // detected, and then merged into the thread they continue. + // RemoteMessageIDsRetired counts older messages whose Google message ID a + // different message reused in the same thread across an ID-space reset; + // each keeps its row under a displaced remote ID instead of being + // overwritten. + IDSpaceResets uint64 `json:"idspace_resets"` + RekeysDeclined uint64 `json:"rekeys_declined"` + RekeysRecovered uint64 `json:"rekeys_recovered"` + RemoteMessageIDsRetired uint64 `json:"remote_message_ids_retired"` } type accountCounters struct { @@ -65,6 +82,11 @@ type accountCounters struct { ephemeral atomic.Uint64 remoteRebinds atomic.Uint64 contentDupesSkipped atomic.Uint64 + + idspaceResets atomic.Uint64 + rekeysDeclined atomic.Uint64 + rekeysRecovered atomic.Uint64 + remoteMessageIDsRetired atomic.Uint64 } // Counters owns atomic ingest counters partitioned by account. Its zero value @@ -149,5 +171,10 @@ func snapshotCounters(c *accountCounters) CounterSnapshot { Ephemeral: c.ephemeral.Load(), RemoteRebinds: c.remoteRebinds.Load(), ContentDupesSkipped: c.contentDupesSkipped.Load(), + + IDSpaceResets: c.idspaceResets.Load(), + RekeysDeclined: c.rekeysDeclined.Load(), + RekeysRecovered: c.rekeysRecovered.Load(), + RemoteMessageIDsRetired: c.remoteMessageIDsRetired.Load(), } } diff --git a/internal/ingest/idspace.go b/internal/ingest/idspace.go index 04afd044..5f4f0ff3 100644 --- a/internal/ingest/idspace.go +++ b/internal/ingest/idspace.go @@ -1,9 +1,13 @@ package ingest import ( + "context" "errors" "fmt" "strconv" + "strings" + "sync" + "time" "github.com/maxghenis/openmessage/internal/bridge" "github.com/maxghenis/openmessage/internal/storage/sqlite" @@ -23,13 +27,112 @@ import ( // The guards here treat participant identity — not the numeric ID — as the // authority on which thread a Google frame belongs to, and migrate the ID // binding whenever the two disagree. +// +// Participant identity alone cannot tell a re-keyed thread from a second live +// thread with the same roster: two groups with the same members, or two 1:1 +// threads with one person, are both live on the phone under their own ids. +// Moving one's binding to the other flips the binding back and forth on every +// event and files both threads' messages into one row. So a roster match may +// take a new id only when its own id is dead (see rekeyable): +// +// - it was displaced (a collision proved its id now names another thread); +// - or the phone announced it (a ConversationEvent reached it) only in an +// earlier device ID space; +// - or it is a provisional thread (minted from a message frame, never +// announced) whose id was bound in an earlier ID space. +// +// The account's remote_idspace_epoch counts detected ID spaces. A new epoch +// starts when a wire id the phone announced for one thread in the current +// epoch arrives naming another: that id was re-keyed, so the phone's ID space +// was reset. Group re-keys also require the event's title to match, so a new +// group never inherits another group's history just because it has the same +// members. +// +// A reset is detected only at the first such collision, and a re-paired phone +// can deliver fresh ids for a few threads before any collision arrives (on +// 2026-09-03, 14 threads in the 36 seconds before the first one). Those +// threads look like second live threads at first and get their own rows. +// When the reset is detected, decisions made in the preceding +// preResetRecoveryWindow are revisited, and a row minted that way is merged +// into the old thread it declined (recoverPreResetDeclines). +// +// Two cases stay ambiguous with the evidence frames carry. After a reset, an +// unannounced id from a peer whose thread has only been re-keyed by messages +// keeps moving that thread until the phone announces it. And a second live +// thread that first appears during a reset's opening minutes is merged into +// its twin. + +// preResetRecoveryWindow is how far back, in frame time, a detected reset +// revisits rows minted because their roster matched a thread that was still +// current. It spans the opening of a re-pair, not ordinary traffic. +const preResetRecoveryWindow = 15 * time.Minute + +// maxRecentDeclines bounds the decision log recoverPreResetDeclines reads. +const maxRecentDeclines = 512 + +// rosterMatch is the identity a frame claims for its thread: a direct thread's +// sole peer, or a group's exact roster and title. +type rosterMatch struct { + direct bool + peers []sqlite.Identity + title string +} + +// rekeyDecline records a frame whose wire id got its own row although its +// roster matched an existing thread, because every match was still current. +type rekeyDecline struct { + accountID string + remoteID string + atMS int64 + match rosterMatch +} + +type declineLog struct { + mu sync.Mutex + entries []rekeyDecline +} + +func (l *declineLog) record(entry rekeyDecline) { + l.mu.Lock() + defer l.mu.Unlock() + cutoff := entry.atMS - preResetRecoveryWindow.Milliseconds() + kept := l.entries[:0] + for _, existing := range l.entries { + if existing.atMS >= cutoff && existing.remoteID != entry.remoteID { + kept = append(kept, existing) + } + } + l.entries = append(kept, entry) + if overflow := len(l.entries) - maxRecentDeclines; overflow > 0 { + l.entries = append([]rekeyDecline(nil), l.entries[overflow:]...) + } +} + +// take removes and returns the account's declines made at or after sinceMS. +func (l *declineLog) take(accountID string, sinceMS int64) []rekeyDecline { + l.mu.Lock() + defer l.mu.Unlock() + var taken []rekeyDecline + kept := l.entries[:0] + for _, entry := range l.entries { + if entry.accountID == accountID && entry.atMS >= sinceMS { + taken = append(taken, entry) + continue + } + kept = append(kept, entry) + } + l.entries = kept + return taken +} // googleIncomingConversation resolves the conversation for an incoming Google -// message whose sender identity is known. The stored binding is used only when -// its direct peer is consistent with the sender; otherwise the binding is -// stale and moves to the sender's own thread (created if absent). An unbound -// ID whose sender already has a direct thread continues that thread rather -// than minting a parallel one. +// message whose sender identity is known. The stored binding is used when its +// direct peer is consistent with the sender. A provisional thread minted in +// the current ID space that hears from a second sender is a group whose +// ConversationEvent has not arrived, and keeps the message. Otherwise a +// contradicting binding is stale and moves to the sender's own thread (created +// if absent). An unbound ID continues the sender's direct thread only when +// that thread's own id is dead; otherwise it is a second live thread. func (w *Worker) googleIncomingConversation( accountID string, platform bridge.Platform, @@ -40,6 +143,11 @@ func (w *Worker) googleIncomingConversation( conversation, remoteID, err := w.existingConversation(accountID, platform, remoteConversationID) if err == nil { if conversation.Kind != sqlite.ConversationKindDirect { + if conversation.Kind == sqlite.ConversationKindGroup { + if err := w.recordProvisionalGroupSender(accountID, conversation, sender); err != nil { + return sqlite.Conversation{}, "", err + } + } return conversation, remoteID, nil } peers, peersErr := w.store.ListConversationPeerIdentities(accountID, conversation.ConversationID) @@ -49,7 +157,26 @@ func (w *Worker) googleIncomingConversation( if len(peers) == 0 || identitiesContain(peers, sender.IdentityID) { return conversation, remoteID, nil } - target, rerouteErr := w.rerouteGoogleDirect(accountID, remoteID, sender, occurredAtMS) + epoch, epochErr := w.store.RemoteIDSpaceEpoch(accountID) + if epochErr != nil { + return sqlite.Conversation{}, "", epochErr + } + binding, bindingErr := w.store.ConversationRemoteBinding(accountID, conversation.ConversationID) + if bindingErr != nil { + return sqlite.Conversation{}, "", bindingErr + } + if binding.Provisional && binding.BoundEpoch >= epoch { + group, adoptErr := w.adoptSecondSender(accountID, conversation, sender) + if adoptErr != nil { + return sqlite.Conversation{}, "", adoptErr + } + return group, remoteID, nil + } + epoch, epochErr = w.noteIDSpaceContradiction(accountID, remoteID, conversation, binding, epoch) + if epochErr != nil { + return sqlite.Conversation{}, "", epochErr + } + target, rerouteErr := w.rerouteGoogleDirect(accountID, remoteID, sender, occurredAtMS, epoch) if rerouteErr != nil { return sqlite.Conversation{}, "", rerouteErr } @@ -59,7 +186,12 @@ func (w *Worker) googleIncomingConversation( return sqlite.Conversation{}, "", err } - target, findErr := w.store.FindDirectConversationBySolePeer(accountID, sender.IdentityID) + epoch, err := w.store.RemoteIDSpaceEpoch(accountID) + if err != nil { + return sqlite.Conversation{}, "", err + } + match := rosterMatch{direct: true, peers: []sqlite.Identity{sender}} + target, findErr := w.findRekeyableThread(accountID, remoteID, match, epoch) if findErr == nil { if bindErr := w.bindRemoteConversationID(accountID, remoteID, &target); bindErr != nil { return sqlite.Conversation{}, "", bindErr @@ -78,15 +210,74 @@ func (w *Worker) googleIncomingConversation( return conversation, remoteID, err } +// adoptSecondSender keeps a second sender's message in a provisional thread +// minted in the current ID space. One wire id names one thread on one phone, +// so a second sender means the thread is a group whose ConversationEvent has +// not arrived (Google can deliver a group's messages before, or without, its +// snapshot). Rerouting instead would scatter the group across its members' +// rows and move the id on every message. The thread becomes a group with the +// new sender as a member, which also stops sole-peer lookups from treating it +// as the first sender's 1:1. +func (w *Worker) adoptSecondSender( + accountID string, + conversation sqlite.Conversation, + sender sqlite.Identity, +) (sqlite.Conversation, error) { + nowMS, err := w.nowMS() + if err != nil { + return sqlite.Conversation{}, err + } + conversation.Kind = sqlite.ConversationKindGroup + conversation.UpdatedAtMS = max(conversation.UpdatedAtMS, nowMS) + if err := w.store.UpsertConversation(conversation); err != nil { + return sqlite.Conversation{}, err + } + if err := w.recordProvisionalGroupSender(accountID, conversation, sender); err != nil { + return sqlite.Conversation{}, err + } + w.logger.Info(). + Str("account_id", accountID). + Str("remote_conversation_id", conversation.RemoteConversationID). + Str("conversation_id", conversation.ConversationID). + Msg("ingest: second sender on a provisional Google thread; treating it as a group until its snapshot arrives") + return w.store.GetConversation(conversation.ConversationID) +} + +// recordProvisionalGroupSender adds a sender to a provisional group's members. +// Until the group's ConversationEvent arrives, its senders are the only roster +// evidence; once announced, the event's roster is authoritative and is left +// alone. +func (w *Worker) recordProvisionalGroupSender( + accountID string, + conversation sqlite.Conversation, + sender sqlite.Identity, +) error { + binding, err := w.store.ConversationRemoteBinding(accountID, conversation.ConversationID) + if err != nil || !binding.Provisional { + return err + } + _, err = w.store.EnsureConversationParticipant(sqlite.ConversationParticipant{ + AccountID: accountID, + ConversationID: conversation.ConversationID, + IdentityID: sender.IdentityID, + Role: sqlite.ParticipantRoleMember, + DisplayName: sender.DisplayName, + IsActive: true, + }) + return err +} + // rerouteGoogleDirect moves a stale direct-thread binding to the sender's own -// thread, minting one when the sender has no direct thread yet. +// thread, minting one when the sender has no direct thread whose id is dead. func (w *Worker) rerouteGoogleDirect( accountID string, remoteID string, sender sqlite.Identity, occurredAtMS int64, + epoch int64, ) (sqlite.Conversation, error) { - target, err := w.store.FindDirectConversationBySolePeer(accountID, sender.IdentityID) + match := rosterMatch{direct: true, peers: []sqlite.Identity{sender}} + target, err := w.findRekeyableThread(accountID, remoteID, match, epoch) if err == nil { if bindErr := w.bindRemoteConversationID(accountID, remoteID, &target); bindErr != nil { return sqlite.Conversation{}, bindErr @@ -126,6 +317,9 @@ func (w *Worker) rerouteGoogleDirect( if err != nil { return sqlite.Conversation{}, err } + if err := w.store.MarkConversationProvisional(accountID, conversation.ConversationID, epoch); err != nil { + return sqlite.Conversation{}, err + } w.counters.account(accountID).remoteRebinds.Add(1) w.logger.Warn(). Str("account_id", accountID). @@ -138,9 +332,11 @@ func (w *Worker) rerouteGoogleDirect( // googleConversationEventTarget picks the row a Google ConversationEvent // should refresh. It returns the stored binding when the event's roster is -// consistent with it, migrates the binding to the roster-matching thread when -// it is not, and resolves an unbound ID to the thread its roster names before -// letting the caller mint a new row (ErrNotFound). +// consistent with it, or when the row is a provisional thread minted in the +// current ID space (the event is the phone's authoritative description of its +// own id). Otherwise it migrates the binding to the roster-matching thread +// whose own id is dead, and resolves an unbound ID the same way before letting +// the caller mint a new row (ErrNotFound). func (w *Worker) googleConversationEventTarget( accountID string, platform bridge.Platform, @@ -158,11 +354,16 @@ func (w *Worker) googleConversationEventTarget( } direct := event.Kind != string(sqlite.ConversationKindGroup) + match := rosterMatch{direct: direct, peers: eventPeers, title: event.Title} if errors.Is(storedErr, sqlite.ErrNotFound) { if len(eventPeers) == 0 { return sqlite.Conversation{}, storedErr } - target, findErr := w.findThreadByRoster(accountID, direct, eventPeers) + epoch, err := w.store.RemoteIDSpaceEpoch(accountID) + if err != nil { + return sqlite.Conversation{}, err + } + target, findErr := w.findRekeyableThread(accountID, remoteID, match, epoch) if errors.Is(findErr, sqlite.ErrNotFound) { return sqlite.Conversation{}, storedErr } @@ -185,10 +386,25 @@ func (w *Worker) googleConversationEventTarget( if len(storedPeers) == 0 || rostersConsistent(direct, eventPeers, storedPeers) { return stored, nil } + epoch, err := w.store.RemoteIDSpaceEpoch(accountID) + if err != nil { + return sqlite.Conversation{}, err + } + binding, err := w.store.ConversationRemoteBinding(accountID, stored.ConversationID) + if err != nil { + return sqlite.Conversation{}, err + } + if binding.Provisional && binding.BoundEpoch >= epoch { + return stored, nil + } // The event's roster names a different thread than the stored binding: // the device-local ID space has been re-keyed under this remote ID. - target, findErr := w.findThreadByRoster(accountID, direct, eventPeers) + epoch, err = w.noteIDSpaceContradiction(accountID, remoteID, stored, binding, epoch) + if err != nil { + return sqlite.Conversation{}, err + } + target, findErr := w.findRekeyableThread(accountID, remoteID, match, epoch) if findErr == nil { if bindErr := w.bindRemoteConversationID(accountID, remoteID, &target); bindErr != nil { return sqlite.Conversation{}, bindErr @@ -298,28 +514,336 @@ func rostersConsistent(direct bool, eventPeers, storedPeers []sqlite.Identity) b return false } -// findThreadByRoster finds the existing thread a roster names: the sole-peer -// direct thread, or the group whose active peer set matches exactly. -func (w *Worker) findThreadByRoster( +// rosterCandidates lists the existing threads a roster names, most recently +// active first: sole-peer direct threads, or groups whose active peer set +// matches exactly and whose title matches the event's. +func (w *Worker) rosterCandidates(accountID string, match rosterMatch) ([]sqlite.Conversation, error) { + if match.direct { + if len(match.peers) != 1 { + return nil, nil + } + return w.store.ListDirectConversationsBySolePeer(accountID, match.peers[0].IdentityID) + } + identityIDs := make([]string, 0, len(match.peers)) + for _, peer := range match.peers { + identityIDs = append(identityIDs, peer.IdentityID) + } + groups, err := w.store.ListGroupConversationsByPeerSet(accountID, identityIDs) + if err != nil { + return nil, err + } + titled := groups[:0] + for _, group := range groups { + if sameThreadTitle(group.Title, match.title) { + titled = append(titled, group) + } + } + return titled, nil +} + +func sameThreadTitle(a, b string) bool { + return strings.EqualFold(strings.Join(strings.Fields(a), " "), strings.Join(strings.Fields(b), " ")) +} + +// findRekeyableThread returns the most recently active thread the roster names +// whose own id is dead (see rekeyable), for remoteID to continue. Matches that +// are still current are other live threads that share the roster; they are +// skipped, and when only such matches exist the result is ErrNotFound so the +// caller gives remoteID its own thread. That decision is logged so a reset +// detected shortly afterwards can revisit it. +func (w *Worker) findRekeyableThread( accountID string, - direct bool, - eventPeers []sqlite.Identity, + remoteID string, + match rosterMatch, + epoch int64, ) (sqlite.Conversation, error) { - if direct { - if len(eventPeers) != 1 { - return sqlite.Conversation{}, fmt.Errorf( - "direct thread roster has %d peers: %w", - len(eventPeers), - sqlite.ErrNotFound, + candidates, err := w.rosterCandidates(accountID, match) + if err != nil { + return sqlite.Conversation{}, err + } + declined := make([]string, 0, len(candidates)) + for _, candidate := range candidates { + if candidate.RemoteConversationID == remoteID { + continue + } + ok, err := w.rekeyable(accountID, candidate, epoch) + if err != nil { + return sqlite.Conversation{}, err + } + if ok { + return candidate, nil + } + declined = append(declined, candidate.RemoteConversationID) + } + if len(declined) > 0 { + nowMS, err := w.nowMS() + if err != nil { + return sqlite.Conversation{}, err + } + w.recentDeclines.record(rekeyDecline{ + accountID: accountID, + remoteID: remoteID, + atMS: nowMS, + match: match, + }) + w.counters.account(accountID).rekeysDeclined.Add(1) + w.logger.Info(). + Str("account_id", accountID). + Str("remote_conversation_id", remoteID). + Strs("live_roster_matches", declined). + Int64("idspace_epoch", epoch). + Msg("ingest: Google thread shares its roster with live threads; keeping it distinct") + } + return sqlite.Conversation{}, fmt.Errorf("re-keyable thread for roster: %w", sqlite.ErrNotFound) +} + +// rekeyable reports whether a roster-matched row may take a different wire id +// because its own id is dead: displaced by a collision, announced by the phone +// only in an earlier ID space, or provisional (never announced) with an id +// bound in an earlier ID space. A row announced in the current ID space is +// live under its own id, and a provisional row bound in the current ID space +// is a new thread awaiting its ConversationEvent (possibly a group whose first +// messages beat its snapshot); neither is taken. +func (w *Worker) rekeyable(accountID string, conversation sqlite.Conversation, epoch int64) (bool, error) { + if isDisplacedConversation(conversation) { + return true, nil + } + binding, err := w.store.ConversationRemoteBinding(accountID, conversation.ConversationID) + if err != nil { + return false, err + } + if binding.Provisional { + return binding.BoundEpoch < epoch, nil + } + return binding.AnnouncedEpoch < epoch, nil +} + +// noteIDSpaceContradiction is called when the row bound to remoteID +// contradicts a frame naming remoteID, and returns the ID-space epoch the +// re-key decision should use. If the phone announced that holder in the +// current ID space, its id has been re-keyed since, so the device ID space +// was reset: the account moves to a new epoch, which makes every thread +// announced only before it re-keyable, and decisions made just before the +// detection are revisited. A holder announced only in an earlier epoch is the +// same reset still being discovered, and a provisional holder proves nothing; +// neither advances the epoch. +func (w *Worker) noteIDSpaceContradiction( + accountID string, + remoteID string, + holder sqlite.Conversation, + binding sqlite.RemoteBinding, + epoch int64, +) (int64, error) { + if isDisplacedConversation(holder) || binding.Provisional || binding.AnnouncedEpoch < epoch { + return epoch, nil + } + next, err := w.store.AdvanceRemoteIDSpaceEpoch(accountID, epoch) + if err != nil { + return 0, err + } + if next == epoch { + return epoch, nil + } + w.counters.account(accountID).idspaceResets.Add(1) + w.logger.Warn(). + Str("account_id", accountID). + Str("remote_conversation_id", remoteID). + Str("conversation_id", holder.ConversationID). + Int64("idspace_epoch", next). + Msg("ingest: Google device ID space was reset (an announced thread id now names another thread)") + if err := w.recoverPreResetDeclines(accountID, next); err != nil { + return 0, err + } + return next, nil +} + +// recoverPreResetDeclines revisits the threads given their own rows in the +// preResetRecoveryWindow before a reset was detected because their roster +// matched a thread that was current then. Under the new epoch that thread's id +// is dead, so the minted row is merged into it and the wire id moves with its +// messages (re-served history that duplicates the old thread is dropped). +func (w *Worker) recoverPreResetDeclines(accountID string, epoch int64) error { + nowMS, err := w.nowMS() + if err != nil { + return err + } + for _, decline := range w.recentDeclines.take(accountID, nowMS-preResetRecoveryWindow.Milliseconds()) { + source, err := w.store.GetConversationByRemote(accountID, decline.remoteID) + if errors.Is(err, sqlite.ErrNotFound) { + continue + } + if err != nil { + return err + } + still, err := w.stillDeclinedShape(accountID, source, decline, epoch) + if err != nil { + return err + } + if !still { + continue + } + candidates, err := w.rosterCandidates(accountID, decline.match) + if err != nil { + return err + } + for _, candidate := range candidates { + if candidate.ConversationID == source.ConversationID { + continue + } + ok, err := w.rekeyable(accountID, candidate, epoch) + if err != nil { + return err + } + if !ok { + continue + } + merge, err := w.store.MergeConversationInto( + context.Background(), + accountID, + decline.remoteID, + source.ConversationID, + candidate.ConversationID, + nowMS, ) + if err != nil { + return err + } + if err := w.store.MarkConversationBound(accountID, candidate.ConversationID, epoch); err != nil { + return err + } + counters := w.counters.account(accountID) + counters.remoteRebinds.Add(1) + counters.rekeysRecovered.Add(1) + w.logger.Warn(). + Str("account_id", accountID). + Str("remote_conversation_id", decline.remoteID). + Str("previous_remote_conversation_id", candidate.RemoteConversationID). + Str("conversation_id", candidate.ConversationID). + Str("merged_conversation_id", source.ConversationID). + Int("moved", merge.Moved). + Int("duplicates", merge.Duplicates). + Int("kept", merge.Kept). + Msg("ingest: Google ID-space reset detected after this thread was kept distinct; merged it into the thread it continues") + break } - return w.store.FindDirectConversationBySolePeer(accountID, eventPeers[0].IdentityID) } - identityIDs := make([]string, 0, len(eventPeers)) - for _, peer := range eventPeers { - identityIDs = append(identityIDs, peer.IdentityID) + return nil +} + +// stillDeclinedShape reports whether the row minted for a declined frame is +// still the thread that decision created: not announced in the new epoch, and +// still the same kind with the same roster. A row that has since become a +// group (more senders) or been announced is left alone. +func (w *Worker) stillDeclinedShape( + accountID string, + source sqlite.Conversation, + decline rekeyDecline, + epoch int64, +) (bool, error) { + binding, err := w.store.ConversationRemoteBinding(accountID, source.ConversationID) + if err != nil { + return false, err + } + if !binding.Provisional && binding.AnnouncedEpoch >= epoch { + return false, nil + } + wantKind := sqlite.ConversationKindGroup + if decline.match.direct { + wantKind = sqlite.ConversationKindDirect } - return w.store.FindGroupConversationByPeerSet(accountID, identityIDs) + if source.Kind != wantKind { + return false, nil + } + peers, err := w.store.ListConversationPeerIdentities(accountID, source.ConversationID) + if err != nil { + return false, err + } + if len(peers) == 0 { + return true, nil + } + if len(peers) != len(decline.match.peers) { + return false, nil + } + for _, peer := range decline.match.peers { + if !identitiesContain(peers, peer.IdentityID) { + return false, nil + } + } + return true, nil +} + +// retireReusedRemoteMessageID frees message's remote ID in its conversation +// when a different message already holds it there. Google message IDs are +// device-local row IDs too, so a thread that continues across an ID-space +// reset can receive a new message under an ID one of its old messages +// carries, and the upsert's natural key would overwrite the old message with +// the new one. The stored row is a different message when it has another +// primary key (it was projected under another wire ID, or is a local outgoing +// placeholder) and differentMessage says so; it then keeps its row under a +// displaced remote ID. +func (w *Worker) retireReusedRemoteMessageID(ctx context.Context, message sqlite.Message) error { + existing, err := w.messages.GetMessageByRemote( + ctx, + message.AccountID, + message.ConversationID, + message.RemoteMessageID, + ) + if errors.Is(err, sqlite.ErrNotFound) { + return nil + } + if err != nil { + return err + } + if existing.MessageID == message.MessageID || !differentMessage(existing, message) { + return nil + } + retired := sqlite.DisplacedRemoteIDPrefix + message.RemoteMessageID + ":" + existing.MessageID + if err := w.messages.RetireRemoteMessageID(ctx, message.AccountID, existing.MessageID, retired); err != nil { + return err + } + w.counters.account(message.AccountID).remoteMessageIDsRetired.Add(1) + w.logger.Warn(). + Str("account_id", message.AccountID). + Str("conversation_id", message.ConversationID). + Str("remote_message_id", message.RemoteMessageID). + Str("retired_message_id", existing.MessageID). + Msg("ingest: Google reused a message ID in a thread that spans two device ID spaces; kept the older message under a displaced ID") + return nil +} + +// reusedMessageIDMinGap separates a reused Google message ID from one message +// seen twice. One message's timestamp moves by seconds between deliveries (an +// outgoing echo replacing its local placeholder, or a re-delivery: at most 16 +// s across the live inbox replayed 2026-10-08), while IDs reused after that +// inbox's 2026-09-03 phone swap named messages 454 to 553 hours apart. +const reusedMessageIDMinGap = time.Hour + +// differentMessage reports whether two messages holding one remote message ID +// in one thread are distinct: opposite directions, or timestamps too far +// apart to be one message. +func differentMessage(stored, incoming sqlite.Message) bool { + if stored.Direction != incoming.Direction { + return true + } + gap := stored.OccurredAtMS - incoming.OccurredAtMS + if gap < 0 { + gap = -gap + } + return gap >= reusedMessageIDMinGap.Milliseconds() +} + +// markGoogleConversationAnnounced records that a ConversationEvent reached the +// conversation in the account's current ID space. +func (w *Worker) markGoogleConversationAnnounced(accountID, conversationID string) error { + epoch, err := w.store.RemoteIDSpaceEpoch(accountID) + if err != nil { + return err + } + return w.store.MarkConversationAnnounced(accountID, conversationID, epoch) +} + +func isDisplacedConversation(conversation sqlite.Conversation) bool { + return strings.HasPrefix(conversation.RemoteConversationID, sqlite.DisplacedRemoteIDPrefix) } // bindRemoteConversationID points an account-scoped remote conversation ID at @@ -347,6 +871,13 @@ func (w *Worker) bindRemoteConversationID( return err } target.RemoteConversationID = remoteID + epoch, err := w.store.RemoteIDSpaceEpoch(accountID) + if err != nil { + return err + } + if err := w.store.MarkConversationBound(accountID, target.ConversationID, epoch); err != nil { + return err + } w.counters.account(accountID).remoteRebinds.Add(1) w.logger.Warn(). Str("account_id", accountID). diff --git a/internal/ingest/idspace_distinct_threads_test.go b/internal/ingest/idspace_distinct_threads_test.go new file mode 100644 index 00000000..e3995aab --- /dev/null +++ b/internal/ingest/idspace_distinct_threads_test.go @@ -0,0 +1,581 @@ +package ingest + +import ( + "errors" + "strconv" + "strings" + "testing" + + "github.com/maxghenis/openmessage/internal/storage/sqlite" +) + +// Two Google threads can share a roster while both are live on the phone: two +// groups with the same members ("Family" and "Book club"), or two 1:1 threads +// with one person. Neither may take the other's v2 row. The #176 roster re-key +// exists for a different case — a device id-space reset, where the old id is +// dead — and the reset tests below pin that it still happens. + +func idsBound(t *testing.T, harness *i01Harness, remoteID string) sqlite.Conversation { + t.Helper() + conversation, err := harness.store.GetConversationByRemote(i01AccountID, remoteID) + if err != nil { + t.Fatalf("GetConversationByRemote(%s): %v", remoteID, err) + } + return conversation +} + +func idsRebinds(harness *i01Harness) uint64 { + return harness.counters.Snapshot(i01AccountID).RemoteRebinds +} + +func idsDisplacedCount(t *testing.T, harness *i01Harness) int64 { + t.Helper() + return i01QueryInt64( + t, + harness.path, + `SELECT COUNT(*) FROM conversations WHERE remote_conversation_id LIKE ?`, + sqlite.DisplacedRemoteIDPrefix+"%", + ) +} + +// idsMessageHome returns the conversation that holds the message with this +// remote message ID (unique across a test's frames). +func idsMessageHome(t *testing.T, harness *i01Harness, remoteMessageID string) string { + t.Helper() + database := i01OpenInspector(t, harness.path) + defer database.Close() + var conversationID string + if err := database.QueryRow( + `SELECT conversation_id FROM messages WHERE remote_message_id = ?`, + remoteMessageID, + ).Scan(&conversationID); err != nil { + t.Fatalf("locate message %s: %v", remoteMessageID, err) + } + return conversationID +} + +func TestGoogleEqualRosterGroupsStayDistinct(t *testing.T) { + script := &idsScript{} + harness := i01NewHarness(t, script, nil) + at := idsAnimalsTimeMS + + script.add("family-conv", idsConversationEvent("5001", "group", "Family", idsKarl, idsShoshana)) + script.add("family-msg", idsIncoming("5001", "1", idsKarl, "family msg", at)) + script.add("book-conv", idsConversationEvent("5002", "group", "Book club", idsKarl, idsShoshana)) + script.add("book-msg", idsIncoming("5002", "2", idsShoshana, "book club msg", at+1000)) + script.add("family-conv-again", idsConversationEvent("5001", "group", "Family", idsKarl, idsShoshana)) + script.add("book-conv-again", idsConversationEvent("5002", "group", "Book club", idsKarl, idsShoshana)) + script.add("family-msg-2", idsIncoming("5001", "3", idsShoshana, "family again", at+2000)) + for _, name := range []string{ + "family-conv", "family-msg", "book-conv", "book-msg", + "family-conv-again", "book-conv-again", "family-msg-2", + } { + idsRun(t, harness, name) + } + + family := idsBound(t, harness, "5001") + book := idsBound(t, harness, "5002") + if family.ConversationID == book.ConversationID { + t.Fatalf("5001 and 5002 share row %q; equal rosters must not merge distinct groups", family.ConversationID) + } + if family.Title != "Family" || book.Title != "Book club" { + t.Fatalf("titles = %q / %q, want Family / Book club", family.Title, book.Title) + } + if got := idsMessageCount(t, harness, family.ConversationID); got != 2 { + t.Fatalf("Family messages = %d, want 2", got) + } + if got := idsMessageCount(t, harness, book.ConversationID); got != 1 { + t.Fatalf("Book club messages = %d, want 1", got) + } + if got := idsRebinds(harness); got != 0 { + t.Fatalf("remote_rebinds = %d, want 0", got) + } + if got := idsDisplacedCount(t, harness); got != 0 { + t.Fatalf("displaced rows = %d, want 0", got) + } +} + +func TestGoogleSameSolePeerDirectThreadsStayDistinct(t *testing.T) { + script := &idsScript{} + harness := i01NewHarness(t, script, nil) + at := idsAnimalsTimeMS + + script.add("sms-conv", idsConversationEvent("6001", "direct", "Karl SMS", idsKarl)) + script.add("sms-msg", idsIncoming("6001", "11", idsKarl, "sms hello", at)) + script.add("rcs-conv", idsConversationEvent("6002", "direct", "Karl RCS", idsKarl)) + script.add("rcs-msg", idsIncoming("6002", "12", idsKarl, "rcs hello", at+1000)) + script.add("sms-conv-again", idsConversationEvent("6001", "direct", "Karl SMS", idsKarl)) + script.add("rcs-conv-again", idsConversationEvent("6002", "direct", "Karl RCS", idsKarl)) + script.add("sms-msg-2", idsIncoming("6001", "13", idsKarl, "sms again", at+2000)) + for _, name := range []string{ + "sms-conv", "sms-msg", "rcs-conv", "rcs-msg", + "sms-conv-again", "rcs-conv-again", "sms-msg-2", + } { + idsRun(t, harness, name) + } + + sms := idsBound(t, harness, "6001") + rcs := idsBound(t, harness, "6002") + if sms.ConversationID == rcs.ConversationID { + t.Fatalf("6001 and 6002 share row %q; one peer's two live threads must stay distinct", sms.ConversationID) + } + if sms.Title != "Karl SMS" || rcs.Title != "Karl RCS" { + t.Fatalf("titles = %q / %q, want Karl SMS / Karl RCS", sms.Title, rcs.Title) + } + if got := idsMessageCount(t, harness, sms.ConversationID); got != 2 { + t.Fatalf("6001 messages = %d, want 2", got) + } + if got := idsMessageCount(t, harness, rcs.ConversationID); got != 1 { + t.Fatalf("6002 messages = %d, want 1", got) + } + for _, conversation := range []sqlite.Conversation{sms, rcs} { + if peers := idsPeerNumbers(t, harness, conversation.ConversationID); len(peers) != 1 || peers[0] != idsKarl { + t.Fatalf("%s peers = %v, want [%s]", conversation.RemoteConversationID, peers, idsKarl) + } + } + if got := idsRebinds(harness); got != 0 { + t.Fatalf("remote_rebinds = %d, want 0", got) + } + if got := idsDisplacedCount(t, harness); got != 0 { + t.Fatalf("displaced rows = %d, want 0", got) + } +} + +func TestGoogleUnboundMessageDoesNotTakeThePeersAnnouncedThread(t *testing.T) { + script := &idsScript{} + harness := i01NewHarness(t, script, nil) + at := idsAnimalsTimeMS + + script.add("sms-conv", idsConversationEvent("6001", "direct", "Karl SMS", idsKarl)) + script.add("sms-msg", idsIncoming("6001", "11", idsKarl, "sms hello", at)) + idsRun(t, harness, "sms-conv") + idsRun(t, harness, "sms-msg") + sms := idsBound(t, harness, "6001") + + // A message under a second live id reaches v2 before that thread's + // ConversationEvent. Karl's announced 6001 thread keeps its id. + script.add("rcs-msg", idsIncoming("6002", "12", idsKarl, "rcs first", at+1000)) + idsRun(t, harness, "rcs-msg") + if got := idsBound(t, harness, "6001"); got.ConversationID != sms.ConversationID { + t.Fatalf("6001 moved to %q, want it to stay on %q", got.ConversationID, sms.ConversationID) + } + rcs := idsBound(t, harness, "6002") + if rcs.ConversationID == sms.ConversationID { + t.Fatal("the 6002 message took Karl's announced 6001 thread") + } + + script.add("rcs-conv", idsConversationEvent("6002", "direct", "Karl RCS", idsKarl)) + script.add("sms-msg-2", idsIncoming("6001", "13", idsKarl, "sms again", at+2000)) + idsRun(t, harness, "rcs-conv") + idsRun(t, harness, "sms-msg-2") + if got := idsBound(t, harness, "6002"); got.ConversationID != rcs.ConversationID || got.Title != "Karl RCS" { + t.Fatalf("6002 after its event = %+v, want row %q titled Karl RCS", got, rcs.ConversationID) + } + if got := idsMessageCount(t, harness, sms.ConversationID); got != 2 { + t.Fatalf("6001 messages = %d, want 2", got) + } + if got := idsMessageCount(t, harness, rcs.ConversationID); got != 1 { + t.Fatalf("6002 messages = %d, want 1", got) + } + if got := idsRebinds(harness); got != 0 { + t.Fatalf("remote_rebinds = %d, want 0", got) + } +} + +func TestGoogleGroupMessageAheadOfItsConversationLeavesOneToOnesIntact(t *testing.T) { + script := &idsScript{} + harness := i01NewHarness(t, script, nil) + at := idsAnimalsTimeMS + + script.add("karl-conv", idsConversationEvent("2916", "direct", "Karl", idsKarl)) + script.add("karl-msg", idsIncoming("2916", "700", idsKarl, "Vote Racine", at)) + script.add("shoshana-conv", idsConversationEvent("3093", "direct", "Shoshana Weissmann", idsShoshana)) + script.add("shoshana-msg", idsIncoming("3093", "86133", idsShoshana, idsAnimalsBody, at)) + for _, name := range []string{"karl-conv", "karl-msg", "shoshana-conv", "shoshana-msg"} { + idsRun(t, harness, name) + } + karl := idsBound(t, harness, "2916") + shoshana := idsBound(t, harness, "3093") + + // A new group's messages arrive before its ConversationEvent. + script.add("group-msg-karl", idsIncoming("4000", "901", idsKarl, "weekend?", at+1000)) + script.add("group-msg-shoshana", idsIncoming("4000", "902", idsShoshana, "saturday works", at+2000)) + script.add("group-conv", idsConversationEvent("4000", "group", "Weekend Group", idsKarl, idsShoshana)) + for _, name := range []string{"group-msg-karl", "group-msg-shoshana", "group-conv"} { + idsRun(t, harness, name) + } + + for _, want := range []struct { + row sqlite.Conversation + remote string + title string + peer string + }{ + {karl, "2916", "Karl", idsKarl}, + {shoshana, "3093", "Shoshana Weissmann", idsShoshana}, + } { + now, err := harness.store.GetConversation(want.row.ConversationID) + if err != nil { + t.Fatalf("GetConversation(%s): %v", want.remote, err) + } + if now.RemoteConversationID != want.remote || now.Kind != sqlite.ConversationKindDirect || now.Title != want.title { + t.Fatalf("1:1 %s after the group's frames = %+v, want it untouched", want.remote, now) + } + if peers := idsPeerNumbers(t, harness, now.ConversationID); len(peers) != 1 || peers[0] != want.peer { + t.Fatalf("1:1 %s peers = %v, want [%s]", want.remote, peers, want.peer) + } + if got := idsMessageCount(t, harness, now.ConversationID); got != 1 { + t.Fatalf("1:1 %s messages = %d, want 1", want.remote, got) + } + } + group := idsBound(t, harness, "4000") + if group.Kind != sqlite.ConversationKindGroup || group.Title != "Weekend Group" { + t.Fatalf("4000 = %+v, want the Weekend Group", group) + } + for _, id := range []string{"901", "902"} { + if home := idsMessageHome(t, harness, id); home != group.ConversationID { + t.Fatalf("message %s filed in %q, want the group %q", id, home, group.ConversationID) + } + } + if got := idsRebinds(harness); got != 0 { + t.Fatalf("remote_rebinds = %d, want 0", got) + } +} + +func TestGoogleIDSpaceResetRekeysStaleThreadsButNotNewOnes(t *testing.T) { + script := &idsScript{} + harness := i01NewHarness(t, script, nil) + _, shoshana, karl := seedOldPhone(t, script, harness) + const g1, g2, g3 = "+15550000001", "+15550000002", "+15550000003" + script.add("old-group", idsConversationEvent("1544", "group", "Giorgio, Mindy, Rohan", g1, g2, g3)) + script.add("old-group-msg", idsIncoming("1544", "900", g1, "dinner?", idsAnimalsTimeMS)) + idsRun(t, harness, "old-group") + idsRun(t, harness, "old-group-msg") + oldGroup := idsBound(t, harness, "1544") + + // New phone. Its id for Shoshana's thread collides with PayPal's old id: + // the reset is detected here. + script.add("new-shoshana", idsConversationEvent("2873", "direct", "Shoshana W.", idsShoshana)) + idsRun(t, harness, "new-shoshana") + if got := idsBound(t, harness, "2873"); got.ConversationID != shoshana.ConversationID { + t.Fatalf("2873 bound to %q, want Shoshana's thread", got.ConversationID) + } + + // Ids that collide with nothing. Karl's thread and the old group were + // announced only by the old phone, so their new ids continue them. + script.add("new-karl", idsConversationEvent("4100", "direct", "Karl", idsKarl)) + script.add("new-group", idsConversationEvent("7000", "group", "Giorgio, Mindy, Rohan", g1, g2, g3)) + idsRun(t, harness, "new-karl") + idsRun(t, harness, "new-group") + if got := idsBound(t, harness, "4100"); got.ConversationID != karl.ConversationID { + t.Fatalf("4100 bound to %q, want Karl's thread %q (re-key after reset)", got.ConversationID, karl.ConversationID) + } + if _, err := harness.store.GetConversationByRemote(i01AccountID, "2916"); !errors.Is(err, sqlite.ErrNotFound) { + t.Fatalf("old id 2916 should be free after the re-key, got err %v", err) + } + if got := idsBound(t, harness, "7000"); got.ConversationID != oldGroup.ConversationID { + t.Fatalf("7000 bound to %q, want the old group %q (re-key after reset)", got.ConversationID, oldGroup.ConversationID) + } + rebindsAfterRekey := idsRebinds(harness) + if rebindsAfterRekey != 3 { + t.Fatalf("remote_rebinds after the re-keys = %d, want 3", rebindsAfterRekey) + } + + // The new phone also has a second group with the same members and a second + // thread with Karl. Their twins were already announced by the new phone, so + // these are different threads. + script.add("new-group-2", idsConversationEvent("7001", "group", "Dinner plans", g1, g2, g3)) + script.add("new-karl-2", idsConversationEvent("4101", "direct", "Karl (work)", idsKarl)) + script.add("new-group-again", idsConversationEvent("7000", "group", "Giorgio, Mindy, Rohan", g1, g2, g3)) + script.add("new-karl-again", idsConversationEvent("4100", "direct", "Karl", idsKarl)) + for _, name := range []string{"new-group-2", "new-karl-2", "new-group-again", "new-karl-again"} { + idsRun(t, harness, name) + } + group2 := idsBound(t, harness, "7001") + if group2.ConversationID == oldGroup.ConversationID { + t.Fatal("7001 took the group the new phone announced as 7000") + } + if got := idsBound(t, harness, "7000"); got.ConversationID != oldGroup.ConversationID || got.Title != "Giorgio, Mindy, Rohan" { + t.Fatalf("7000 after its twin appeared = %+v, want the old group, title unchanged", got) + } + if group2.Title != "Dinner plans" { + t.Fatalf("7001 title = %q, want Dinner plans", group2.Title) + } + karl2 := idsBound(t, harness, "4101") + if karl2.ConversationID == karl.ConversationID { + t.Fatal("4101 took the Karl thread the new phone announced as 4100") + } + if got := idsBound(t, harness, "4100"); got.ConversationID != karl.ConversationID { + t.Fatalf("4100 moved to %q, want Karl's thread", got.ConversationID) + } + if got := idsRebinds(harness); got != rebindsAfterRekey { + t.Fatalf("remote_rebinds = %d, want still %d", got, rebindsAfterRekey) + } + if strings.HasPrefix(group2.RemoteConversationID, sqlite.DisplacedRemoteIDPrefix) { + t.Fatalf("7001 row displaced: %+v", group2) + } +} + +// The id-space state lives in the store: a worker restarted after a reset +// still treats old-phone threads as re-keyable and new-phone threads as live. +func TestGoogleIDSpaceStateSurvivesWorkerRestart(t *testing.T) { + script := &idsScript{} + harness := i01NewHarness(t, script, nil) + _, shoshana, karl := seedOldPhone(t, script, harness) + script.add("new-shoshana", idsConversationEvent("2873", "direct", "Shoshana W.", idsShoshana)) + idsRun(t, harness, "new-shoshana") + + counters := &Counters{} + restarted := &i01Harness{ + path: harness.path, + store: harness.store, + messages: harness.messages, + counters: counters, + } + restarted.worker = i01NewWorker(t, harness.store, harness.messages, counters, script, nil) + restarted.sink = i01NewSink(t, harness.messages, restarted.worker, counters, "inbox-restart") + + script.add("new-karl", idsConversationEvent("4100", "direct", "Karl", idsKarl)) + script.add("new-shoshana-2", idsConversationEvent("4200", "direct", "Shoshana (2)", idsShoshana)) + idsRun(t, restarted, "new-karl") + idsRun(t, restarted, "new-shoshana-2") + if got := idsBound(t, restarted, "4100"); got.ConversationID != karl.ConversationID { + t.Fatalf("after restart, 4100 bound to %q, want Karl's old-phone thread re-keyed", got.ConversationID) + } + if got := idsBound(t, restarted, "4200"); got.ConversationID == shoshana.ConversationID { + t.Fatal("after restart, 4200 took Shoshana's thread the new phone announced as 2873") + } + if got := idsBound(t, restarted, "2873"); got.ConversationID != shoshana.ConversationID { + t.Fatalf("2873 moved to %q", got.ConversationID) + } +} + +// A row minted from a message frame takes the message's own timestamp as its +// creation time. When the phone's clock runs ahead of this one, the thread's +// ConversationEvent used to fail the updated_at_ms >= created_at_ms check and +// was quarantined, leaving the thread untitled and roster-less. +func TestGoogleConversationEventAfterFutureDatedMessageIsApplied(t *testing.T) { + script := &idsScript{} + harness := i01NewHarness(t, script, nil) + ahead := i01TestTime.UnixMilli() + 5_000 + script.add("msg", idsIncoming("9100", "1", idsKarl, "from a fast phone clock", ahead)) + script.add("conv", idsConversationEvent("9100", "direct", "Karl", idsKarl)) + idsRun(t, harness, "msg") + idsRun(t, harness, "conv") + got := idsBound(t, harness, "9100") + if got.Title != "Karl" { + t.Fatalf("9100 title = %q, want the ConversationEvent's Karl", got.Title) + } + if got.UpdatedAtMS < got.CreatedAtMS { + t.Fatalf("9100 updated_at %d < created_at %d", got.UpdatedAtMS, got.CreatedAtMS) + } +} + +// Replays the 2026-09-03 re-pair: the new phone delivered messages under +// fresh ids for 14 threads in the 36 seconds before the first id collision +// revealed the reset. Each looked like a second live thread with its peer and +// got its own row; once the reset is detected those rows fold back into the +// threads they continue, and re-served history is not kept twice. +func TestGoogleResetDetectedLateMergesThreadsKeptDistinctBeforeIt(t *testing.T) { + script := &idsScript{} + harness := i01NewHarness(t, script, nil) + _, shoshana, karl := seedOldPhone(t, script, harness) + + script.add("new-karl-reserved", idsIncoming("4400", "40", idsKarl, "Vote Racine", idsAnimalsTimeMS-3_600_000)) + script.add("new-karl-live", idsIncoming("4400", "41", idsKarl, "polls close at 8", idsAnimalsTimeMS+60_000)) + idsRun(t, harness, "new-karl-reserved") + idsRun(t, harness, "new-karl-live") + early := idsBound(t, harness, "4400") + if early.ConversationID == karl.ConversationID { + t.Fatal("before any reset evidence, 4400 took Karl's announced thread") + } + + // The first collision: the new phone's id for Shoshana is PayPal's old id. + script.add("new-shoshana-replay", idsIncoming("2873", "59", idsShoshana, idsAnimalsBody, idsAnimalsTimeMS)) + idsRun(t, harness, "new-shoshana-replay") + + if got := idsBound(t, harness, "4400"); got.ConversationID != karl.ConversationID { + t.Fatalf("after the reset was detected, 4400 bound to %q, want Karl's thread %q", got.ConversationID, karl.ConversationID) + } + if got := idsMessageCount(t, harness, karl.ConversationID); got != 2 { + t.Fatalf("Karl thread messages = %d, want 2 (original + new; the re-served copy dropped)", got) + } + if home := idsMessageHome(t, harness, "41"); home != karl.ConversationID { + t.Fatalf("new message 41 in %q, want Karl's thread", home) + } + if _, err := harness.store.GetConversation(early.ConversationID); !errors.Is(err, sqlite.ErrNotFound) { + t.Fatalf("emptied interim row still present: %v", err) + } + if got := idsBound(t, harness, "2873"); got.ConversationID != shoshana.ConversationID { + t.Fatalf("2873 bound to %q, want Shoshana's thread", got.ConversationID) + } + snapshot := harness.counters.Snapshot(i01AccountID) + if snapshot.IDSpaceResets != 1 || snapshot.RekeysRecovered != 1 { + t.Fatalf("idspace_resets %d rekeys_recovered %d, want 1/1", snapshot.IDSpaceResets, snapshot.RekeysRecovered) + } + + script.add("new-karl-later", idsIncoming("4400", "42", idsKarl, "thanks", idsAnimalsTimeMS+120_000)) + idsRun(t, harness, "new-karl-later") + if home := idsMessageHome(t, harness, "42"); home != karl.ConversationID { + t.Fatalf("later 4400 message in %q, want Karl's thread", home) + } +} + +// Replays wire 3097 in the live inbox: a group's messages from several +// members arrive with no ConversationEvent. Rerouting each message by sender +// scattered the group across per-sender fragments and moved the id on every +// message (15 times in 5 seconds on main). +func TestGoogleGroupWithoutItsConversationEventStaysOneThread(t *testing.T) { + script := &idsScript{} + harness := i01NewHarness(t, script, nil) + at := idsAnimalsTimeMS + const u = "+15550000001" + for _, seed := range []struct{ remote, title, peer string }{ + {"2916", "Karl", idsKarl}, + {"3093", "Shoshana Weissmann", idsShoshana}, + {"3200", "U", u}, + } { + script.add("conv-"+seed.remote, idsConversationEvent(seed.remote, "direct", seed.title, seed.peer)) + idsRun(t, harness, "conv-"+seed.remote) + } + senders := []string{idsKarl, idsShoshana, u, idsShoshana, idsKarl, u, idsKarl} + for index, sender := range senders { + name := "group-msg-" + strconv.Itoa(index) + script.add(name, idsIncoming("3097", strconv.Itoa(900+index), sender, "group line "+strconv.Itoa(index), at+int64(index)*1000)) + idsRun(t, harness, name) + } + group := idsBound(t, harness, "3097") + if got := idsMessageCount(t, harness, group.ConversationID); got != int64(len(senders)) { + t.Fatalf("3097 thread messages = %d, want all %d", got, len(senders)) + } + if group.Kind != sqlite.ConversationKindGroup { + t.Fatalf("3097 kind = %q, want group once a second sender appeared", group.Kind) + } + if peers := idsPeerNumbers(t, harness, group.ConversationID); len(peers) != 3 { + t.Fatalf("3097 members = %v, want all three senders", peers) + } + if got := idsRebinds(harness); got != 0 { + t.Fatalf("remote_rebinds = %d, want 0", got) + } + if got := idsDisplacedCount(t, harness); got != 0 { + t.Fatalf("displaced rows = %d, want 0", got) + } + for _, remote := range []string{"2916", "3093", "3200"} { + if got := idsMessageCount(t, harness, idsBound(t, harness, remote).ConversationID); got != 0 { + t.Fatalf("1:1 %s took %d group messages", remote, got) + } + } + + script.add("group-conv", idsConversationEvent("3097", "group", "Weekend", idsKarl, idsShoshana, u)) + idsRun(t, harness, "group-conv") + if got := idsBound(t, harness, "3097"); got.ConversationID != group.ConversationID || got.Title != "Weekend" { + t.Fatalf("3097 after its event = %+v, want the same row titled Weekend", got) + } +} + +// After a reset, groups with one roster re-key by title: each continues its +// own history, and a new group with that roster continues neither. +func TestGoogleResetRekeysSameRosterGroupsByTitle(t *testing.T) { + script := &idsScript{} + harness := i01NewHarness(t, script, nil) + const q = "+15550000004" + at := idsAnimalsTimeMS + script.add("family", idsConversationEvent("5001", "group", "Family", idsKarl, q)) + script.add("family-msg", idsIncoming("5001", "930", idsKarl, "Family old", at+100)) + script.add("book", idsConversationEvent("5002", "group", "Book club", idsKarl, q)) + script.add("book-msg", idsIncoming("5002", "931", idsKarl, "Book old", at+200)) + for _, name := range []string{"family", "family-msg", "book", "book-msg"} { + idsRun(t, harness, name) + } + family := idsBound(t, harness, "5001") + book := idsBound(t, harness, "5002") + _, _, _ = seedOldPhone(t, script, harness) + script.add("reset", idsConversationEvent("2873", "direct", "Shoshana W.", idsShoshana)) + idsRun(t, harness, "reset") + + script.add("new-family", idsConversationEvent("7001", "group", "Family", idsKarl, q)) + script.add("new-book", idsConversationEvent("7002", "group", "Book club", idsKarl, q)) + script.add("new-club", idsConversationEvent("7003", "group", "New club", idsKarl, q)) + for _, name := range []string{"new-family", "new-book", "new-club"} { + idsRun(t, harness, name) + } + if got := idsBound(t, harness, "7001"); got.ConversationID != family.ConversationID { + t.Fatalf("7001 Family bound to %q, want the Family thread %q", got.ConversationID, family.ConversationID) + } + if got := idsBound(t, harness, "7002"); got.ConversationID != book.ConversationID { + t.Fatalf("7002 Book club bound to %q, want the Book club thread %q", got.ConversationID, book.ConversationID) + } + club := idsBound(t, harness, "7003") + if club.ConversationID == family.ConversationID || club.ConversationID == book.ConversationID { + t.Fatal("a new group with the same members took an old group's history") + } + if got := idsMessageCount(t, harness, club.ConversationID); got != 0 { + t.Fatalf("new club inherited %d messages", got) + } +} + +// A thread that never got a ConversationEvent (minted from its messages) on +// the old phone still re-keys after a reset, and its re-served history dedupes. +func TestGoogleResetRekeysOldPhoneProvisionalThreads(t *testing.T) { + script := &idsScript{} + harness := i01NewHarness(t, script, nil) + script.add("paypal", idsConversationEvent("2873", "direct", "72975", idsPayPal)) + script.add("shoshana-msg", idsIncoming("3093", "86133", idsShoshana, idsAnimalsBody, idsAnimalsTimeMS)) + idsRun(t, harness, "paypal") + idsRun(t, harness, "shoshana-msg") + shoshana := idsBound(t, harness, "3093") + + script.add("reset", idsIncoming("2873", "59", idsShoshana, idsAnimalsBody, idsAnimalsTimeMS)) + idsRun(t, harness, "reset") + if got := idsBound(t, harness, "2873"); got.ConversationID != shoshana.ConversationID { + t.Fatalf("2873 bound to %q, want Shoshana's message-minted thread %q", got.ConversationID, shoshana.ConversationID) + } + if got := idsMessageCount(t, harness, shoshana.ConversationID); got != 1 { + t.Fatalf("Shoshana thread messages = %d, want 1 (re-served copy deduped)", got) + } +} + +// Google message ids are device-local too. A thread that continues across a +// reset can receive a new message under an id one of its old messages holds; +// the old message must survive. +func TestGoogleReusedMessageIDInAContinuedThreadKeepsBothMessages(t *testing.T) { + script := &idsScript{} + harness := i01NewHarness(t, script, nil) + _, _, karl := seedOldPhone(t, script, harness) + script.add("reset", idsConversationEvent("2873", "direct", "Shoshana W.", idsShoshana)) + script.add("new-karl", idsConversationEvent("4100", "direct", "Karl", idsKarl)) + idsRun(t, harness, "reset") + idsRun(t, harness, "new-karl") + if got := idsBound(t, harness, "4100"); got.ConversationID != karl.ConversationID { + t.Fatalf("4100 bound to %q, want Karl's thread", got.ConversationID) + } + + later := idsAnimalsTimeMS + 30*24*3_600_000 + script.add("reused-id", idsIncoming("4100", "700", idsKarl, "a different message", later)) + idsRun(t, harness, "reused-id") + if got := idsMessageCount(t, harness, karl.ConversationID); got != 2 { + t.Fatalf("Karl thread messages = %d, want both the old and the new message 700", got) + } + var oldBody string + database := i01OpenInspector(t, harness.path) + defer database.Close() + if err := database.QueryRow( + `SELECT body FROM messages WHERE conversation_id = ? AND remote_message_id LIKE ?`, + karl.ConversationID, sqlite.DisplacedRemoteIDPrefix+"700:%", + ).Scan(&oldBody); err != nil || oldBody != "Vote Racine" { + t.Fatalf("old message 700 = %q (err %v), want it kept under a displaced id", oldBody, err) + } + if got := harness.counters.Snapshot(i01AccountID).RemoteMessageIDsRetired; got != 1 { + t.Fatalf("remote_message_ids_retired = %d, want 1", got) + } + + // The same new message re-delivered seconds later is one message. + script.add("reused-id-again", idsIncoming("4100", "700", idsKarl, "a different message", later+5_000)) + idsRun(t, harness, "reused-id-again") + if got := idsMessageCount(t, harness, karl.ConversationID); got != 2 { + t.Fatalf("after a re-delivery, Karl thread messages = %d, want 2", got) + } + if got := harness.counters.Snapshot(i01AccountID).RemoteMessageIDsRetired; got != 1 { + t.Fatalf("re-delivery retired another id: %d", got) + } +} diff --git a/internal/ingest/idspace_properties_test.go b/internal/ingest/idspace_properties_test.go new file mode 100644 index 00000000..a33f7a54 --- /dev/null +++ b/internal/ingest/idspace_properties_test.go @@ -0,0 +1,341 @@ +package ingest + +import ( + "fmt" + "math/rand/v2" + "os" + "strconv" + "testing" + + "github.com/maxghenis/openmessage/internal/bridge" +) + +// Property tests for Google wire-id binding. Each case generates a random +// phone: threads with wire ids, kinds, titles and rosters, deliberately +// reusing rosters so several live threads share one. Frames are fed through a +// real worker and store. +// +// Invariants (no id-space reset): +// N1. Every announced thread keeps the row its wire id was bound to when it +// was first announced, for every later frame. +// N2. Announced threads never share a row. +// N3. Each announced thread's row carries that thread's title. +// N4. A message sent to an announced thread lands in that thread's row. +// N5. With each thread announced before its first message, no frame rebinds +// or displaces anything. +// N6. No message is lost: every message fed is stored in some thread. +// Invariants (one id-space reset, then the new phone lists its threads twice): +// R1. After the listings, every new-phone thread has its own row, titled for +// that thread, and messages land there. +// R2. The second listing changes no binding and rebinds nothing. +// R3. When the very first new-phone frame collides with a different old-phone +// thread (so the reset is detected before any fresh id is seen), every +// thread whose roster is unique continues its old row. +// R4. No message is lost. +// +// Direct peers and group members come from disjoint pools, and two group +// rosters are either identical or disjoint. Overlapping-but-unequal rosters +// are treated as one evolving group by design (rostersConsistent), which these +// properties do not model. + +var ( + propDirectPeers = []string{"+15551000001", "+15551000002", "+15551000003"} + propGroupRosters = [][]string{ + {"+15552000001", "+15552000002"}, + {"+15552000003", "+15552000004", "+15552000005"}, + } +) + +type propThread struct { + remoteID string + kind string + title string + roster []string +} + +func (p propThread) conversationEvent() bridge.Event { + return idsConversationEvent(p.remoteID, p.kind, p.title, p.roster...) +} + +func (p propThread) rosterKey() string { + return p.kind + fmt.Sprint(p.roster) +} + +func propCases() int { + if value, err := strconv.Atoi(os.Getenv("OPENMESSAGE_PROPERTY_CASES")); err == nil && value > 0 { + return value + } + return 60 +} + +func propRandomThread(rng *rand.Rand, remoteID, title string) propThread { + if rng.IntN(2) == 0 { + return propThread{ + remoteID: remoteID, + kind: "direct", + title: title, + roster: []string{propDirectPeers[rng.IntN(len(propDirectPeers))]}, + } + } + return propThread{ + remoteID: remoteID, + kind: "group", + title: title, + roster: propGroupRosters[rng.IntN(len(propGroupRosters))], + } +} + +type propRunner struct { + t *testing.T + script *idsScript + harness *i01Harness + frames int + nextMsg int + clockMS int64 +} + +func newPropRunner(t *testing.T) *propRunner { + script := &idsScript{} + return &propRunner{ + t: t, + script: script, + harness: i01NewHarness(t, script, nil), + clockMS: idsAnimalsTimeMS, + } +} + +func (r *propRunner) feed(events ...bridge.Event) { + r.t.Helper() + r.frames++ + name := "prop-" + strconv.Itoa(r.frames) + r.script.add(name, events...) + idsRun(r.t, r.harness, name) +} + +// message feeds one incoming message from a roster member and returns its +// remote message id. +func (r *propRunner) message(rng *rand.Rand, thread propThread) string { + r.t.Helper() + r.nextMsg++ + r.clockMS += 1000 + id := "m" + strconv.Itoa(r.nextMsg) + sender := thread.roster[rng.IntN(len(thread.roster))] + r.feed(idsIncoming(thread.remoteID, id, sender, "body "+id, r.clockMS)) + return id +} + +func (r *propRunner) assertNoMessageLost(label string) { + r.t.Helper() + stored := i01QueryInt64(r.t, r.harness.path, `SELECT COUNT(*) FROM messages`) + if stored != int64(r.nextMsg) { + r.t.Fatalf("%s: %d messages stored, want all %d fed", label, stored, r.nextMsg) + } +} + +func (r *propRunner) boundRow(remoteID string) (string, string, bool) { + conversation, err := r.harness.store.GetConversationByRemote(i01AccountID, remoteID) + if err != nil { + return "", "", false + } + return conversation.ConversationID, conversation.Title, true +} + +func TestGooglePropertyLiveThreadsKeepTheirRows(t *testing.T) { + for seed := uint64(1); seed <= uint64(propCases()); seed++ { + t.Run(fmt.Sprintf("seed=%d", seed), func(t *testing.T) { + rng := rand.New(rand.NewPCG(seed, 0x6f6d)) + announceFirst := seed%2 == 1 + threads := make([]propThread, 2+rng.IntN(5)) + for i := range threads { + threads[i] = propRandomThread(rng, strconv.Itoa(8000+i), "thread "+strconv.Itoa(i)) + } + r := newPropRunner(t) + announcedRow := map[string]string{} + messageThread := map[string]string{} + + check := func(step string) { + t.Helper() + owners := map[string]string{} + for _, thread := range threads { + want, announced := announcedRow[thread.remoteID] + if !announced { + continue + } + row, title, ok := r.boundRow(thread.remoteID) + if !ok || row != want { + t.Fatalf("%s: N1 %s bound to %q (bound=%v), want %q", step, thread.remoteID, row, ok, want) + } + if other, taken := owners[row]; taken { + t.Fatalf("%s: N2 %s and %s share row %q", step, other, thread.remoteID, row) + } + owners[row] = thread.remoteID + if title != thread.title { + t.Fatalf("%s: N3 %s title %q, want %q", step, thread.remoteID, title, thread.title) + } + } + for id, remoteID := range messageThread { + if home := idsMessageHome(t, r.harness, id); home != announcedRow[remoteID] { + t.Fatalf("%s: N4 message %s in %q, want %s's row %q", step, id, home, remoteID, announcedRow[remoteID]) + } + } + } + + steps := len(threads) * (2 + rng.IntN(4)) + for step := 0; step < steps; step++ { + thread := threads[rng.IntN(len(threads))] + _, announced := announcedRow[thread.remoteID] + if (announceFirst && !announced) || rng.IntN(3) == 0 { + r.feed(thread.conversationEvent()) + if !announced { + row, _, ok := r.boundRow(thread.remoteID) + if !ok { + t.Fatalf("step %d: %s unbound right after its event", step, thread.remoteID) + } + announcedRow[thread.remoteID] = row + } + } else { + id := r.message(rng, thread) + if announced { + messageThread[id] = thread.remoteID + } + } + check(fmt.Sprintf("step %d", step)) + } + for _, thread := range threads { + r.feed(thread.conversationEvent()) + if _, announced := announcedRow[thread.remoteID]; !announced { + row, _, _ := r.boundRow(thread.remoteID) + announcedRow[thread.remoteID] = row + } + } + check("final listing") + r.assertNoMessageLost("N6") + if announceFirst { + if got := idsRebinds(r.harness); got != 0 { + t.Fatalf("N5 remote_rebinds = %d, want 0", got) + } + if got := idsDisplacedCount(t, r.harness); got != 0 { + t.Fatalf("N5 displaced rows = %d, want 0", got) + } + } + }) + } +} + +func TestGooglePropertyResetRekeysAndSettles(t *testing.T) { + for seed := uint64(1); seed <= uint64(propCases()); seed++ { + t.Run(fmt.Sprintf("seed=%d", seed), func(t *testing.T) { + rng := rand.New(rand.NewPCG(seed, 0x7265)) + count := 2 + rng.IntN(5) + + // Old phone: ids 100..; each thread announced with one message. + old := make([]propThread, count) + for i := range old { + old[i] = propRandomThread(rng, strconv.Itoa(100+i), "thread "+strconv.Itoa(i)) + } + r := newPropRunner(t) + oldRow := make([]string, count) + for i, thread := range old { + r.feed(thread.conversationEvent()) + r.message(rng, thread) + oldRow[i], _, _ = r.boundRow(thread.remoteID) + } + + // New phone: the same threads under fresh ids drawn from a range + // that overlaps the old ids, plus possibly one extra thread whose + // roster twins an existing one. + pool := rng.Perm(2 * count) + fresh := make([]propThread, count) + for i, thread := range old { + thread.remoteID = strconv.Itoa(100 + pool[i]) + fresh[i] = thread + } + if rng.IntN(2) == 0 { + twin := fresh[rng.IntN(count)] + twin.remoteID = strconv.Itoa(100 + pool[count]) + twin.title = "twin of " + twin.title + fresh = append(fresh, twin) + } + rosterUses := map[string]int{} + for _, thread := range fresh { + rosterUses[thread.rosterKey()]++ + } + + order := rng.Perm(len(fresh)) + first := fresh[order[0]] + detectedFirst := false + for i, thread := range old { + if thread.remoteID == first.remoteID && first.rosterKey() != thread.rosterKey() { + detectedFirst = true + _ = i + } + } + + for _, index := range order { + r.feed(fresh[index].conversationEvent()) + } + for _, index := range rng.Perm(len(fresh)) { + r.feed(fresh[index].conversationEvent()) + } + rows := map[string]string{} + owners := map[string]string{} + for _, thread := range fresh { + row, title, ok := r.boundRow(thread.remoteID) + if !ok { + t.Fatalf("R1 %s unbound after two listings", thread.remoteID) + } + if other, taken := owners[row]; taken { + t.Fatalf("R1 %s and %s share row %q", other, thread.remoteID, row) + } + owners[row] = thread.remoteID + if title != thread.title { + t.Fatalf("R1 %s title %q, want %q", thread.remoteID, title, thread.title) + } + rows[thread.remoteID] = row + } + + rebinds := idsRebinds(r.harness) + for _, index := range rng.Perm(len(fresh)) { + r.feed(fresh[index].conversationEvent()) + } + if got := idsRebinds(r.harness); got != rebinds { + t.Fatalf("R2 third listing rebinds = %d, want %d", got, rebinds) + } + for _, thread := range fresh { + if row, _, _ := r.boundRow(thread.remoteID); row != rows[thread.remoteID] { + t.Fatalf("R2 %s moved from %q to %q on re-listing", thread.remoteID, rows[thread.remoteID], row) + } + } + + for _, thread := range fresh { + id := r.message(rng, thread) + if home := idsMessageHome(t, r.harness, id); home != rows[thread.remoteID] { + t.Fatalf("R1 message %s in %q, want %s's row %q", id, home, thread.remoteID, rows[thread.remoteID]) + } + } + + r.assertNoMessageLost("R4") + if detectedFirst { + for i := range old { + thread := fresh[i] + if rosterUses[thread.rosterKey()] != 1 { + continue + } + oldRoster := 0 + for _, o := range old { + if o.rosterKey() == thread.rosterKey() { + oldRoster++ + } + } + if oldRoster != 1 { + continue + } + if rows[thread.remoteID] != oldRow[i] { + t.Fatalf("R3 %s (old id %s) bound to %q, want its old row %q", + thread.remoteID, old[i].remoteID, rows[thread.remoteID], oldRow[i]) + } + } + } + }) + } +} diff --git a/internal/ingest/worker.go b/internal/ingest/worker.go index ee120ef4..a4b9782d 100644 --- a/internal/ingest/worker.go +++ b/internal/ingest/worker.go @@ -80,6 +80,9 @@ type Worker struct { changeMu sync.Mutex changed chan struct{} + + // recentDeclines feeds recoverPreResetDeclines (see idspace.go). + recentDeclines declineLog } // NewWorker validates and copies its explicitly configured decoder @@ -484,6 +487,11 @@ func (w *Worker) applyEvents( } continue } + if platform == bridge.PlatformGoogle { + if err := w.retireReusedRemoteMessageID(ctx, projection.Message); err != nil { + return false, err + } + } if messageCount == 0 { if err := w.messages.ProjectMessage(ctx, projection); err != nil { return false, err @@ -831,7 +839,11 @@ func (w *Worker) refreshConversation( } conversation.Title = event.Title conversation.RemoteRevision = optionalTrimmed(event.RemoteRevision) - conversation.UpdatedAtMS = nowMS + // Never move updated_at backwards: a row minted from a message frame + // carries the phone's timestamp, which can lead this clock when the + // phone's clock runs ahead, and the schema requires updated_at_ms >= + // created_at_ms. + conversation.UpdatedAtMS = max(conversation.UpdatedAtMS, nowMS) } if err := w.store.UpsertConversation(conversation); err != nil { return sqlite.Conversation{}, err @@ -859,6 +871,13 @@ func (w *Worker) refreshConversation( if err := w.store.ReplaceConversationParticipants(conversation.ConversationID, participants); err != nil { return sqlite.Conversation{}, err } + if platform == bridge.PlatformGoogle { + // The phone just announced this thread under remoteID: it is live in + // the current device ID space (see idspace.go). + if err := w.markGoogleConversationAnnounced(accountID, conversation.ConversationID); err != nil { + return sqlite.Conversation{}, err + } + } return conversation, nil } @@ -907,7 +926,21 @@ func (w *Worker) ensureMessageConversation( return sqlite.Conversation{}, "", err } conversation, err = w.store.GetConversationByRemote(accountID, remoteID) - return conversation, remoteID, err + if err != nil { + return sqlite.Conversation{}, "", err + } + if platform == bridge.PlatformGoogle { + // No ConversationEvent has described this thread yet: it is + // provisional in the current device ID space (see idspace.go). + epoch, err := w.store.RemoteIDSpaceEpoch(accountID) + if err != nil { + return sqlite.Conversation{}, "", err + } + if err := w.store.MarkConversationProvisional(accountID, conversation.ConversationID, epoch); err != nil { + return sqlite.Conversation{}, "", err + } + } + return conversation, remoteID, nil } func (w *Worker) existingConversation( diff --git a/internal/migration/transform_test.go b/internal/migration/transform_test.go index de61a879..e6e7c80f 100644 --- a/internal/migration/transform_test.go +++ b/internal/migration/transform_test.go @@ -522,8 +522,8 @@ func assertFixtureReport(t *testing.T, report Report, sourceHash string) { t.Errorf("source file evidence did not reconcile: %+v", file) } } - if report.Target.SchemaVersion != 10 || len(report.Target.MigrationChecksums) != 10 { - t.Fatalf("target schema = version %d with %d checksums, want version 10 with 10 checksums", report.Target.SchemaVersion, len(report.Target.MigrationChecksums)) + if report.Target.SchemaVersion != 11 || len(report.Target.MigrationChecksums) != 11 { + t.Fatalf("target schema = version %d with %d checksums, want version 11 with 11 checksums", report.Target.SchemaVersion, len(report.Target.MigrationChecksums)) } wantTargetCounts := map[string]int64{ "accounts": 5, "devices": 5, "people": 1, "person_identities": 2, diff --git a/internal/migration/validate.go b/internal/migration/validate.go index 4f7465e9..73af1aaf 100644 --- a/internal/migration/validate.go +++ b/internal/migration/validate.go @@ -23,7 +23,10 @@ var targetCountTables = []string{ "read_cursors", } -const migration0010Checksum = "dfab4551335d92045cb895e5b2c781f4f57216bbc428a705fc26bbdd474db0b1" +const ( + migration0010Checksum = "dfab4551335d92045cb895e5b2c781f4f57216bbc428a705fc26bbdd474db0b1" + migration0011Checksum = "cc3515064aaa221866dcc011d1efca62c1604047a4701be898d4422ad2b1a42a" +) func checkpointAndSyncSQLite(ctx context.Context, path string) error { database, err := sql.Open("sqlite", path) @@ -151,8 +154,9 @@ func validateTarget( countsMatched := countsMatch(dataset, state, report, actualHistory, actualScheduled, actualHistoryByPlatform) report.Validation.CountsMatched = countsMatched report.Validation.Passed = quick == "ok" && - report.Target.SchemaVersion == 10 && len(report.Target.MigrationChecksums) == 10 && + report.Target.SchemaVersion == 11 && len(report.Target.MigrationChecksums) == 11 && report.Target.MigrationChecksums[9] == migration0010Checksum && + report.Target.MigrationChecksums[10] == migration0011Checksum && len(fkViolations) == 0 && orphanTotal(orphans) == 0 && countsMatched && report.Validation.SampledHashesMatched && report.Validation.BlobReferencesValid && report.Validation.SourceUnchanged && diff --git a/internal/storage/sqlite/attachments_test.go b/internal/storage/sqlite/attachments_test.go index eb49522f..165f2534 100644 --- a/internal/storage/sqlite/attachments_test.go +++ b/internal/storage/sqlite/attachments_test.go @@ -322,8 +322,8 @@ func openAttachmentTestRepository(t *testing.T) (*Store, *AttachmentRepository) } }) - if len(embeddedMigrations) != 10 { - t.Fatalf("embedded migrations = %d, want 10", len(embeddedMigrations)) + if len(embeddedMigrations) != 11 { + t.Fatalf("embedded migrations = %d, want 11", len(embeddedMigrations)) } assertPragmaInt(t, store.db, "user_version", len(embeddedMigrations)) ledger := readLedgerRow(t, store.db, 3) diff --git a/internal/storage/sqlite/identity_graph_test.go b/internal/storage/sqlite/identity_graph_test.go index 1f8bb81a..37afcecb 100644 --- a/internal/storage/sqlite/identity_graph_test.go +++ b/internal/storage/sqlite/identity_graph_test.go @@ -162,8 +162,8 @@ func openIdentityGraphTestStore(t *testing.T) *Store { } }) - if len(embeddedMigrations) != 10 { - t.Fatalf("embedded migrations = %d, want 10", len(embeddedMigrations)) + if len(embeddedMigrations) != 11 { + t.Fatalf("embedded migrations = %d, want 11", len(embeddedMigrations)) } assertPragmaInt(t, store.db, "user_version", len(embeddedMigrations)) ledger := readLedgerRow(t, store.db, 2) diff --git a/internal/storage/sqlite/message_attachments_test.go b/internal/storage/sqlite/message_attachments_test.go index 5ccd12a2..102fcf85 100644 --- a/internal/storage/sqlite/message_attachments_test.go +++ b/internal/storage/sqlite/message_attachments_test.go @@ -83,8 +83,8 @@ func TestMessageAttachmentsMigrationAppliesToBlankAndExistingV7AndReopens(t *tes } }) after := readLedgerRows(t, store.db) - if len(after) != 10 { - t.Fatalf("migrated ledger rows = %d, want 10", len(after)) + if len(after) != 11 { + t.Fatalf("migrated ledger rows = %d, want 11", len(after)) } if !slices.Equal(after[:7], before) { t.Fatalf("migrations 0001-0007 changed:\nbefore: %+v\nafter: %+v", before, after[:7]) @@ -252,10 +252,10 @@ func TestMessageAttachmentRowsCascadeWithMessageDeletion(t *testing.T) { func assertMessageAttachmentsMigration(t *testing.T, store *Store) { t.Helper() - if len(embeddedMigrations) != 10 { - t.Fatalf("embedded migrations = %d, want 10", len(embeddedMigrations)) + if len(embeddedMigrations) != 11 { + t.Fatalf("embedded migrations = %d, want 11", len(embeddedMigrations)) } - assertPragmaInt(t, store.db, "user_version", 10) + assertPragmaInt(t, store.db, "user_version", 11) ledger := readLedgerRow(t, store.db, 8) if ledger.name != "message_attachments" { t.Fatalf("migration 0008 name = %q, want message_attachments", ledger.name) diff --git a/internal/storage/sqlite/messages.go b/internal/storage/sqlite/messages.go index e2bffd0c..65c5c097 100644 --- a/internal/storage/sqlite/messages.go +++ b/internal/storage/sqlite/messages.go @@ -257,6 +257,39 @@ func (r *MessageRepository) GetMessage( return message, nil } +// RetireRemoteMessageID renames a message's remote message ID to retiredID so +// the original ID is free for a different message. Google reuses message IDs +// across device ID spaces, and a thread that continues across a phone swap can +// receive a new message under an ID one of its old messages already holds; the +// old message keeps its row, reactions and attachments under the retired ID. +func (r *MessageRepository) RetireRemoteMessageID( + ctx context.Context, + accountID string, + messageID string, + retiredID string, +) error { + if strings.TrimSpace(retiredID) == "" { + return fmt.Errorf("retire remote ID of message %q: retired ID is empty", messageID) + } + nowMS := r.now().UnixMilli() + result, err := r.store.db.ExecContext(ctx, ` + UPDATE messages + SET remote_message_id = ?, updated_at_ms = MAX(updated_at_ms, ?) + WHERE account_id = ? AND message_id = ? + `, retiredID, nowMS, accountID, messageID) + if err != nil { + return fmt.Errorf("retire remote ID of message %q: %w", messageID, mapConstraintError(err)) + } + affected, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("retire remote ID of message %q: rows affected: %w", messageID, err) + } + if affected != 1 { + return fmt.Errorf("retire remote ID of message %q: %w", messageID, ErrNotFound) + } + return nil +} + // GetMessageByRemote returns the normalized message selected by the same // account-scoped natural key used by ProjectMessage. This is useful to recover // the effective local message ID after a projection collides with a row that diff --git a/internal/storage/sqlite/messages_test.go b/internal/storage/sqlite/messages_test.go index 70449ac5..4841b510 100644 --- a/internal/storage/sqlite/messages_test.go +++ b/internal/storage/sqlite/messages_test.go @@ -1178,8 +1178,8 @@ func TestMessagesInboxMigrationIsChecksummedAndStrict(t *testing.T) { t, func() time.Time { return time.UnixMilli(messageTestTimeMS) }, ) - if len(embeddedMigrations) != 10 { - t.Fatalf("embedded migrations = %d, want 10", len(embeddedMigrations)) + if len(embeddedMigrations) != 11 { + t.Fatalf("embedded migrations = %d, want 11", len(embeddedMigrations)) } assertPragmaInt(t, store.db, "user_version", len(embeddedMigrations)) ledger := readLedgerRow(t, store.db, 4) diff --git a/internal/storage/sqlite/migrations.go b/internal/storage/sqlite/migrations.go index ba26da57..c6cbca30 100644 --- a/internal/storage/sqlite/migrations.go +++ b/internal/storage/sqlite/migrations.go @@ -66,6 +66,9 @@ var migration0009SQL string //go:embed migrations/0010_reactions.sql var migration0010SQL string +//go:embed migrations/0011_remote_idspace_epochs.sql +var migration0011SQL string + var embeddedMigrations = []migration{ newMigration(1, "storage_shell", migration0001SQL, newStorageShellArguments), newMigration(2, "identity_graph", migration0002SQL, nil), @@ -77,6 +80,7 @@ var embeddedMigrations = []migration{ newMigration(8, "message_attachments", migration0008SQL, nil), newMigration(9, "outbox_send_again", migration0009SQL, nil), newMigration(10, "reactions", migration0010SQL, nil), + newMigration(11, "remote_idspace_epochs", migration0011SQL, nil), } func newMigration( diff --git a/internal/storage/sqlite/migrations/0011_remote_idspace_epochs.sql b/internal/storage/sqlite/migrations/0011_remote_idspace_epochs.sql new file mode 100644 index 00000000..b0d21503 --- /dev/null +++ b/internal/storage/sqlite/migrations/0011_remote_idspace_epochs.sql @@ -0,0 +1,26 @@ +-- Google Messages conversation ids are device-local row ids: a phone swap, +-- factory reset or backup restore re-keys every thread (a new "ID space"), so +-- ingest re-binds ids by participant identity. Two live threads can also share +-- a roster (two groups with the same members, two 1:1 threads with one +-- person), and those must never take each other's row. These columns let +-- ingest tell the two apart. +-- +-- accounts.remote_idspace_epoch counts the device ID spaces ingest has +-- detected for the account. +-- +-- conversations.remote_announced_epoch is the latest epoch in which the +-- transport announced the conversation (a ConversationEvent was applied to the +-- row). NULL marks a provisional thread: ingest minted it from a message frame +-- and no ConversationEvent has reached it yet. Rows written by anything else +-- (pre-existing rows, legacy migration, mirroring, repair) default to epoch 0. +-- +-- conversations.remote_bound_epoch is the epoch in which the row's current +-- wire id was bound to it. +ALTER TABLE accounts ADD COLUMN remote_idspace_epoch INTEGER NOT NULL DEFAULT 0 + CHECK (remote_idspace_epoch >= 0); + +ALTER TABLE conversations ADD COLUMN remote_announced_epoch INTEGER DEFAULT 0 + CHECK (remote_announced_epoch IS NULL OR remote_announced_epoch >= 0); + +ALTER TABLE conversations ADD COLUMN remote_bound_epoch INTEGER NOT NULL DEFAULT 0 + CHECK (remote_bound_epoch >= 0); diff --git a/internal/storage/sqlite/outbox_test.go b/internal/storage/sqlite/outbox_test.go index a4dea9c6..5f4a879f 100644 --- a/internal/storage/sqlite/outbox_test.go +++ b/internal/storage/sqlite/outbox_test.go @@ -2473,10 +2473,10 @@ func TestOutboxMigrationIsChecksummedAndStrict(t *testing.T) { store, _ := openOutboxTestRepository(t, func() time.Time { return time.UnixMilli(outboxTestTimeMS) }) - if len(embeddedMigrations) != 10 { - t.Fatalf("embedded migrations = %d, want 10", len(embeddedMigrations)) + if len(embeddedMigrations) != 11 { + t.Fatalf("embedded migrations = %d, want 11", len(embeddedMigrations)) } - assertPragmaInt(t, store.db, "user_version", 10) + assertPragmaInt(t, store.db, "user_version", 11) ledger := readLedgerRow(t, store.db, 5) if ledger.name != "outbox" { t.Fatalf("migration 0005 name = %q, want outbox", ledger.name) @@ -2583,8 +2583,8 @@ func TestOutboxSendAgainMigrationAppliesToBlankAndExistingV8DatabaseWithRows(t * } }) after := readLedgerRows(t, store.db) - if len(after) != 10 { - t.Fatalf("migrated ledger rows = %d, want 10", len(after)) + if len(after) != 11 { + t.Fatalf("migrated ledger rows = %d, want 11", len(after)) } if !slices.Equal(after[:8], before) { t.Fatalf("migrations 0001-0008 changed:\nbefore: %+v\nafter: %+v", before, after[:8]) @@ -2613,7 +2613,7 @@ func TestOutboxSendAgainMigrationAppliesToBlankAndExistingV8DatabaseWithRows(t * func assertOutboxSendAgainMigration(t *testing.T, store *Store) { t.Helper() - assertPragmaInt(t, store.db, "user_version", 10) + assertPragmaInt(t, store.db, "user_version", 11) ledger := readLedgerRow(t, store.db, 9) if ledger.name != "outbox_send_again" { t.Fatalf("migration 0009 name = %q, want outbox_send_again", ledger.name) @@ -2715,8 +2715,8 @@ func TestOutboxAttachmentsMigrationAppliesToBlankAndExistingV5Database(t *testin } }) after := readLedgerRows(t, store.db) - if len(after) != 10 { - t.Fatalf("migrated ledger rows = %d, want 10", len(after)) + if len(after) != 11 { + t.Fatalf("migrated ledger rows = %d, want 11", len(after)) } if !slices.Equal(after[:5], before) { t.Fatalf("migrations 0001-0005 changed:\nbefore: %+v\nafter: %+v", before, after[:5]) @@ -2727,7 +2727,7 @@ func TestOutboxAttachmentsMigrationAppliesToBlankAndExistingV5Database(t *testin func assertOutboxAttachmentsMigration(t *testing.T, store *Store) { t.Helper() - assertPragmaInt(t, store.db, "user_version", 10) + assertPragmaInt(t, store.db, "user_version", 11) ledger := readLedgerRow(t, store.db, 6) if ledger.name != "outbox_attachments" { t.Fatalf("migration 0006 name = %q, want outbox_attachments", ledger.name) @@ -2782,7 +2782,7 @@ func TestReactionsReadMigrationAppliesToBlankAndReopens(t *testing.T) { func assertReactionsReadMigration(t *testing.T, store *Store) { t.Helper() - assertPragmaInt(t, store.db, "user_version", 10) + assertPragmaInt(t, store.db, "user_version", 11) ledger := readLedgerRow(t, store.db, 7) if ledger.name != "reactions_read" { t.Fatalf("migration 0007 name = %q, want reactions_read", ledger.name) diff --git a/internal/storage/sqlite/reactions_migration_test.go b/internal/storage/sqlite/reactions_migration_test.go index 6f925307..af6eb84d 100644 --- a/internal/storage/sqlite/reactions_migration_test.go +++ b/internal/storage/sqlite/reactions_migration_test.go @@ -65,8 +65,8 @@ func TestReactionsMigrationAppliesToBlankAndExistingV9Database(t *testing.T) { } }) after := readLedgerRows(t, store.db) - if len(after) != 10 { - t.Fatalf("migrated ledger rows = %d, want 10", len(after)) + if len(after) != 11 { + t.Fatalf("migrated ledger rows = %d, want 11", len(after)) } if !slices.Equal(after[:9], before) { t.Fatalf("migrations 0001-0009 changed:\nbefore: %+v\nafter: %+v", before, after[:9]) @@ -77,10 +77,10 @@ func TestReactionsMigrationAppliesToBlankAndExistingV9Database(t *testing.T) { func assertReactionsMigration(t *testing.T, store *Store) { t.Helper() - if len(embeddedMigrations) != 10 { - t.Fatalf("embedded migrations = %d, want 10", len(embeddedMigrations)) + if len(embeddedMigrations) != 11 { + t.Fatalf("embedded migrations = %d, want 11", len(embeddedMigrations)) } - assertPragmaInt(t, store.db, "user_version", 10) + assertPragmaInt(t, store.db, "user_version", 11) ledger := readLedgerRow(t, store.db, 10) if ledger.name != "reactions" { t.Fatalf("migration 0010 name = %q, want reactions", ledger.name) diff --git a/internal/storage/sqlite/rebind.go b/internal/storage/sqlite/rebind.go index 3cea6bf4..a3fc1d51 100644 --- a/internal/storage/sqlite/rebind.go +++ b/internal/storage/sqlite/rebind.go @@ -71,7 +71,28 @@ func (s *Store) FindDirectConversationBySolePeer( accountID string, identityID string, ) (Conversation, error) { - conversation, err := scanConversation(s.db.QueryRowContext(context.Background(), ` + conversations, err := s.ListDirectConversationsBySolePeer(accountID, identityID) + if err != nil { + return Conversation{}, err + } + if len(conversations) == 0 { + return Conversation{}, fmt.Errorf( + "direct conversation with sole peer %q: %w", + identityID, + ErrNotFound, + ) + } + return conversations[0], nil +} + +// ListDirectConversationsBySolePeer returns every direct conversation whose +// only active non-self participant is the given identity, most recently active +// first. Displaced rows are included. +func (s *Store) ListDirectConversationsBySolePeer( + accountID string, + identityID string, +) ([]Conversation, error) { + rows, err := s.db.QueryContext(context.Background(), ` SELECT `+conversationColumns+` FROM conversations c WHERE c.account_id = ? AND c.kind = 'direct' @@ -88,23 +109,23 @@ func (s *Store) FindDirectConversationBySolePeer( AND p2.is_active = 1 AND i2.is_self = 0 AND p2.identity_id <> ? ) ORDER BY c.last_message_at_ms DESC, c.conversation_id - LIMIT 1 - `, accountID, identityID, identityID)) - if errors.Is(err, sql.ErrNoRows) { - return Conversation{}, fmt.Errorf( - "direct conversation with sole peer %q: %w", + `, accountID, identityID, identityID) + if err != nil { + return nil, fmt.Errorf( + "list direct conversations with sole peer %q: %w", identityID, - ErrNotFound, + err, ) } + conversations, err := collectRows(rows, scanConversation) if err != nil { - return Conversation{}, fmt.Errorf( - "find direct conversation with sole peer %q: %w", + return nil, fmt.Errorf( + "list direct conversations with sole peer %q: %w", identityID, err, ) } - return conversation, nil + return conversations, nil } // FindGroupConversationByPeerSet returns the group conversation whose active @@ -114,8 +135,25 @@ func (s *Store) FindGroupConversationByPeerSet( accountID string, identityIDs []string, ) (Conversation, error) { + groups, err := s.ListGroupConversationsByPeerSet(accountID, identityIDs) + if err != nil { + return Conversation{}, err + } + if len(groups) == 0 { + return Conversation{}, fmt.Errorf("group conversation with given peer set: %w", ErrNotFound) + } + return groups[0], nil +} + +// ListGroupConversationsByPeerSet returns every group conversation whose +// active non-self participant identities are exactly the given set, most +// recently active first. Displaced rows are included. +func (s *Store) ListGroupConversationsByPeerSet( + accountID string, + identityIDs []string, +) ([]Conversation, error) { if len(identityIDs) == 0 { - return Conversation{}, fmt.Errorf("group conversation peer set is empty: %w", ErrNotFound) + return nil, nil } want := make(map[string]struct{}, len(identityIDs)) for _, id := range identityIDs { @@ -129,16 +167,17 @@ func (s *Store) FindGroupConversationByPeerSet( ORDER BY c.last_message_at_ms DESC, c.conversation_id `, accountID) if err != nil { - return Conversation{}, fmt.Errorf("list group conversations: %w", err) + return nil, fmt.Errorf("list group conversations: %w", err) } groups, err := collectRows(rows, scanConversation) if err != nil { - return Conversation{}, fmt.Errorf("list group conversations: %w", err) + return nil, fmt.Errorf("list group conversations: %w", err) } + matches := make([]Conversation, 0, 1) for _, group := range groups { peers, err := s.ListConversationPeerIdentities(accountID, group.ConversationID) if err != nil { - return Conversation{}, err + return nil, err } if len(peers) != len(want) { continue @@ -151,10 +190,321 @@ func (s *Store) FindGroupConversationByPeerSet( } } if match { - return group, nil + matches = append(matches, group) + } + } + return matches, nil +} + +// RemoteIDSpaceEpoch returns the account's current device ID-space epoch: the +// number of remote ID-space resets ingest has detected for it. +func (s *Store) RemoteIDSpaceEpoch(accountID string) (int64, error) { + var epoch int64 + err := s.db.QueryRowContext(context.Background(), ` + SELECT remote_idspace_epoch FROM accounts WHERE account_id = ? + `, accountID).Scan(&epoch) + if errors.Is(err, sql.ErrNoRows) { + return 0, fmt.Errorf("remote ID-space epoch for account %q: %w", accountID, ErrNotFound) + } + if err != nil { + return 0, fmt.Errorf("read remote ID-space epoch for account %q: %w", accountID, err) + } + return epoch, nil +} + +// AdvanceRemoteIDSpaceEpoch moves the account from epoch from to from+1 and +// returns the account's epoch afterwards. When the account has already moved +// past from, it returns that newer epoch without advancing again. +func (s *Store) AdvanceRemoteIDSpaceEpoch(accountID string, from int64) (int64, error) { + if _, err := s.db.ExecContext(context.Background(), ` + UPDATE accounts + SET remote_idspace_epoch = remote_idspace_epoch + 1 + WHERE account_id = ? AND remote_idspace_epoch = ? + `, accountID, from); err != nil { + return 0, fmt.Errorf("advance remote ID-space epoch for account %q: %w", accountID, err) + } + return s.RemoteIDSpaceEpoch(accountID) +} + +// RemoteBinding is a conversation's device ID-space provenance (see migration +// 0011). +type RemoteBinding struct { + // Provisional is true for a thread ingest minted from a message frame that + // no ConversationEvent has reached yet. + Provisional bool + // AnnouncedEpoch is the latest ID-space epoch in which a ConversationEvent + // was applied to the row. Zero when Provisional. + AnnouncedEpoch int64 + // BoundEpoch is the epoch in which the row's current wire id was bound. + BoundEpoch int64 +} + +// ConversationRemoteBinding returns the conversation's ID-space provenance. +func (s *Store) ConversationRemoteBinding(accountID, conversationID string) (RemoteBinding, error) { + var announced sql.NullInt64 + var binding RemoteBinding + err := s.db.QueryRowContext(context.Background(), ` + SELECT remote_announced_epoch, remote_bound_epoch FROM conversations + WHERE account_id = ? AND conversation_id = ? + `, accountID, conversationID).Scan(&announced, &binding.BoundEpoch) + if errors.Is(err, sql.ErrNoRows) { + return RemoteBinding{}, fmt.Errorf("remote binding of conversation %q: %w", conversationID, ErrNotFound) + } + if err != nil { + return RemoteBinding{}, fmt.Errorf("read remote binding of conversation %q: %w", conversationID, err) + } + binding.Provisional = !announced.Valid + binding.AnnouncedEpoch = announced.Int64 + return binding, nil +} + +// MarkConversationAnnounced records that the transport announced the +// conversation (a ConversationEvent was applied to it) in the given ID-space +// epoch, which also confirms its current binding in that epoch. Neither epoch +// moves backwards, and the row stops being provisional. +func (s *Store) MarkConversationAnnounced(accountID, conversationID string, epoch int64) error { + return s.updateRemoteBinding("announced", ` + UPDATE conversations + SET remote_announced_epoch = MAX(COALESCE(remote_announced_epoch, 0), ?1), + remote_bound_epoch = MAX(remote_bound_epoch, ?1) + WHERE account_id = ?2 AND conversation_id = ?3 + `, accountID, conversationID, epoch) +} + +// MarkConversationProvisional records that ingest minted the conversation from +// a message frame in the given epoch: it is provisional until a +// ConversationEvent announces it. +func (s *Store) MarkConversationProvisional(accountID, conversationID string, epoch int64) error { + return s.updateRemoteBinding("provisional", ` + UPDATE conversations + SET remote_announced_epoch = NULL, remote_bound_epoch = ?1 + WHERE account_id = ?2 AND conversation_id = ?3 + `, accountID, conversationID, epoch) +} + +// MarkConversationBound records that the conversation's current wire id was +// bound to it in the given epoch. It never moves backwards. +func (s *Store) MarkConversationBound(accountID, conversationID string, epoch int64) error { + return s.updateRemoteBinding("bound", ` + UPDATE conversations + SET remote_bound_epoch = MAX(remote_bound_epoch, ?1) + WHERE account_id = ?2 AND conversation_id = ?3 + `, accountID, conversationID, epoch) +} + +func (s *Store) updateRemoteBinding( + what string, + statement string, + accountID string, + conversationID string, + epoch int64, +) error { + if epoch < 0 { + return fmt.Errorf("mark conversation %q %s: epoch %d is negative", conversationID, what, epoch) + } + result, err := s.db.ExecContext(context.Background(), statement, epoch, accountID, conversationID) + if err != nil { + return fmt.Errorf("mark conversation %q %s: %w", conversationID, what, err) + } + affected, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("mark conversation %q %s: rows affected: %w", conversationID, what, err) + } + if affected != 1 { + return fmt.Errorf("mark conversation %q %s: %w", conversationID, what, ErrNotFound) + } + return nil +} + +// ConversationMerge reports what MergeConversationInto did. +type ConversationMerge struct { + // Moved messages now live in the target conversation. + Moved int + // Duplicates were deleted because the target already held the same + // message (direction, sender, millisecond and body) under another remote id. + Duplicates int + // Kept messages stayed behind because the target already uses their remote + // message id for a different message. + Kept int + // Dropped is true when the emptied source conversation was deleted. + Dropped bool +} + +// MergeConversationInto folds the conversation bound to remoteID into target +// and binds remoteID to target, in one transaction. It is the undo for a +// thread minted under a re-keyed wire id before the ID-space reset that +// re-keyed it was detected: each message moves to target, except a content +// duplicate of a target message (deleted) or one whose remote message id the +// target already uses (left behind). The source keeps whatever is left under a +// displaced marker, and is deleted when nothing references it. +func (s *Store) MergeConversationInto( + ctx context.Context, + accountID string, + remoteID string, + sourceID string, + targetID string, + nowMS int64, +) (ConversationMerge, error) { + var merge ConversationMerge + if sourceID == targetID { + return merge, fmt.Errorf("merge conversation %q into itself", sourceID) + } + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return merge, fmt.Errorf("merge conversation %q: begin: %w", sourceID, err) + } + defer tx.Rollback() + + var holderID string + err = tx.QueryRowContext(ctx, ` + SELECT conversation_id FROM conversations + WHERE account_id = ? AND remote_conversation_id = ? + `, accountID, remoteID).Scan(&holderID) + if errors.Is(err, sql.ErrNoRows) || (err == nil && holderID != sourceID) { + return merge, fmt.Errorf("merge conversation %q: %q is no longer bound to it: %w", sourceID, remoteID, ErrNotFound) + } + if err != nil { + return merge, fmt.Errorf("merge conversation %q: read holder: %w", sourceID, err) + } + var targetExists int + if err := tx.QueryRowContext(ctx, ` + SELECT COUNT(*) FROM conversations WHERE account_id = ? AND conversation_id = ? + `, accountID, targetID).Scan(&targetExists); err != nil { + return merge, fmt.Errorf("merge conversation %q: read target: %w", sourceID, err) + } + if targetExists != 1 { + return merge, fmt.Errorf("merge conversation %q: target %q: %w", sourceID, targetID, ErrNotFound) + } + + // Read cursors point at messages of their own conversation; the source's + // cursors cannot follow its messages. + if _, err := tx.ExecContext(ctx, ` + DELETE FROM read_cursors WHERE account_id = ? AND conversation_id = ? + `, accountID, sourceID); err != nil { + return merge, fmt.Errorf("merge conversation %q: drop read cursors: %w", sourceID, err) + } + + type sourceMessage struct { + messageID, remoteMessageID, direction, body string + senderIdentityID sql.NullString + occurredAtMS int64 + } + rows, err := tx.QueryContext(ctx, ` + SELECT message_id, remote_message_id, direction, body, sender_identity_id, occurred_at_ms + FROM messages WHERE account_id = ? AND conversation_id = ? + ORDER BY occurred_at_ms, message_id + `, accountID, sourceID) + if err != nil { + return merge, fmt.Errorf("merge conversation %q: list messages: %w", sourceID, err) + } + var messages []sourceMessage + for rows.Next() { + var message sourceMessage + if err := rows.Scan( + &message.messageID, + &message.remoteMessageID, + &message.direction, + &message.body, + &message.senderIdentityID, + &message.occurredAtMS, + ); err != nil { + rows.Close() + return merge, fmt.Errorf("merge conversation %q: scan message: %w", sourceID, err) + } + messages = append(messages, message) + } + if err := rows.Close(); err != nil { + return merge, fmt.Errorf("merge conversation %q: list messages: %w", sourceID, err) + } + + for _, message := range messages { + if strings.TrimSpace(message.body) != "" { + var duplicate int + if err := tx.QueryRowContext(ctx, ` + SELECT COUNT(*) FROM messages + WHERE account_id = ? AND conversation_id = ? AND occurred_at_ms = ? + AND direction = ? AND sender_identity_id IS ? AND body = ? + AND remote_message_id <> ? + `, accountID, targetID, message.occurredAtMS, message.direction, + message.senderIdentityID, message.body, message.remoteMessageID).Scan(&duplicate); err != nil { + return merge, fmt.Errorf("merge conversation %q: match duplicate: %w", sourceID, err) + } + if duplicate > 0 { + if _, err := tx.ExecContext(ctx, ` + DELETE FROM messages WHERE account_id = ? AND message_id = ? + `, accountID, message.messageID); err != nil { + return merge, fmt.Errorf("merge conversation %q: delete duplicate: %w", sourceID, err) + } + merge.Duplicates++ + continue + } + } + var collision int + if err := tx.QueryRowContext(ctx, ` + SELECT COUNT(*) FROM messages + WHERE account_id = ? AND conversation_id = ? AND remote_message_id = ? + `, accountID, targetID, message.remoteMessageID).Scan(&collision); err != nil { + return merge, fmt.Errorf("merge conversation %q: match remote id: %w", sourceID, err) + } + if collision > 0 { + merge.Kept++ + continue + } + if _, err := tx.ExecContext(ctx, ` + UPDATE reactions SET conversation_id = ?, updated_at_ms = MAX(updated_at_ms, ?) + WHERE account_id = ? AND message_id = ? + `, targetID, nowMS, accountID, message.messageID); err != nil { + return merge, fmt.Errorf("merge conversation %q: move reactions: %w", sourceID, err) + } + if _, err := tx.ExecContext(ctx, ` + UPDATE messages SET conversation_id = ?, updated_at_ms = MAX(updated_at_ms, ?) + WHERE account_id = ? AND message_id = ? + `, targetID, nowMS, accountID, message.messageID); err != nil { + return merge, fmt.Errorf("merge conversation %q: move message: %w", sourceID, err) } + merge.Moved++ + } + + if _, err := tx.ExecContext(ctx, ` + UPDATE conversations + SET remote_conversation_id = ?, updated_at_ms = MAX(updated_at_ms, ?) + WHERE account_id = ? AND conversation_id = ? + `, displacedRemoteID(remoteID, sourceID), nowMS, accountID, sourceID); err != nil { + return merge, fmt.Errorf("merge conversation %q: displace: %w", sourceID, err) + } + if _, err := tx.ExecContext(ctx, ` + UPDATE conversations + SET remote_conversation_id = ?, + last_message_at_ms = MAX(last_message_at_ms, COALESCE(( + SELECT MAX(occurred_at_ms) FROM messages + WHERE account_id = conversations.account_id + AND conversation_id = conversations.conversation_id + ), 0)), + updated_at_ms = MAX(updated_at_ms, ?) + WHERE account_id = ? AND conversation_id = ? + `, remoteID, nowMS, accountID, targetID); err != nil { + return merge, fmt.Errorf("merge conversation %q: bind target %q: %w", sourceID, targetID, err) + } + if merge.Kept == 0 { + result, err := tx.ExecContext(ctx, ` + DELETE FROM conversations + WHERE account_id = ? AND conversation_id = ? + AND NOT EXISTS (SELECT 1 FROM messages WHERE conversation_id = conversations.conversation_id) + AND NOT EXISTS (SELECT 1 FROM outbox WHERE conversation_id = conversations.conversation_id) + `, accountID, sourceID) + if err != nil { + return merge, fmt.Errorf("merge conversation %q: drop: %w", sourceID, err) + } + affected, err := result.RowsAffected() + if err != nil { + return merge, fmt.Errorf("merge conversation %q: drop: rows affected: %w", sourceID, err) + } + merge.Dropped = affected == 1 + } + if err := tx.Commit(); err != nil { + return merge, fmt.Errorf("merge conversation %q: commit: %w", sourceID, err) } - return Conversation{}, fmt.Errorf("group conversation with given peer set: %w", ErrNotFound) + return merge, nil } // ReassignConversationRemoteID moves an account-scoped remote conversation ID diff --git a/internal/storage/sqlite/rebind_test.go b/internal/storage/sqlite/rebind_test.go new file mode 100644 index 00000000..cb25a841 --- /dev/null +++ b/internal/storage/sqlite/rebind_test.go @@ -0,0 +1,337 @@ +package sqlite + +import ( + "context" + "database/sql" + "errors" + "path/filepath" + "slices" + "testing" +) + +const rebindTestTimeMS = int64(1_790_000_000_000) + +func seedRebindFixture(t *testing.T, db *sql.DB) { + t.Helper() + mustExec(t, db, ` + INSERT INTO accounts (account_id, bridge_key, created_at_ms, updated_at_ms) + VALUES ('google', 'google_messages', ?, ?) + `, rebindTestTimeMS, rebindTestTimeMS) + mustExec(t, db, ` + INSERT INTO identities (identity_id, account_id, kind, canonical_value, raw_value, created_at_ms, updated_at_ms) + VALUES + ('karl', 'google', 'e164', '+12026022529', '+12026022529', ?, ?), + ('shoshana', 'google', 'e164', '+15169021075', '+15169021075', ?, ?) + `, rebindTestTimeMS, rebindTestTimeMS, rebindTestTimeMS, rebindTestTimeMS) + mustExec(t, db, ` + INSERT INTO conversations (conversation_id, account_id, remote_conversation_id, kind, title, last_message_at_ms, created_at_ms, updated_at_ms) + VALUES + ('karl-sms', 'google', '6001', 'direct', 'Karl SMS', 100, ?, ?), + ('karl-rcs', 'google', '6002', 'direct', 'Karl RCS', 300, ?, ?), + ('karl-old', 'google', 'displaced:2916:karl-old', 'direct', 'Karl', 200, ?, ?), + ('family', 'google', '5001', 'group', 'Family', 100, ?, ?), + ('book-club', 'google', '5002', 'group', 'Book club', 200, ?, ?), + ('trio', 'google', '5003', 'group', 'Trio', 300, ?, ?) + `, + rebindTestTimeMS, rebindTestTimeMS, rebindTestTimeMS, rebindTestTimeMS, + rebindTestTimeMS, rebindTestTimeMS, rebindTestTimeMS, rebindTestTimeMS, + rebindTestTimeMS, rebindTestTimeMS, rebindTestTimeMS, rebindTestTimeMS, + ) + mustExec(t, db, ` + INSERT INTO conversation_participants (account_id, conversation_id, identity_id) + VALUES + ('google', 'karl-sms', 'karl'), + ('google', 'karl-rcs', 'karl'), + ('google', 'karl-old', 'karl'), + ('google', 'family', 'karl'), + ('google', 'family', 'shoshana'), + ('google', 'book-club', 'karl'), + ('google', 'book-club', 'shoshana'), + ('google', 'trio', 'karl') + `) +} + +func conversationIDs(conversations []Conversation) []string { + ids := make([]string, 0, len(conversations)) + for _, conversation := range conversations { + ids = append(ids, conversation.ConversationID) + } + return ids +} + +func TestRosterCandidateListsReturnEveryMatchMostRecentFirst(t *testing.T) { + store := openIdentityGraphTestStore(t) + seedRebindFixture(t, store.db) + + directs, err := store.ListDirectConversationsBySolePeer("google", "karl") + if err != nil { + t.Fatalf("ListDirectConversationsBySolePeer(): %v", err) + } + if got, want := conversationIDs(directs), []string{"karl-rcs", "karl-old", "karl-sms"}; !slices.Equal(got, want) { + t.Fatalf("direct candidates = %v, want %v (displaced included, most recent first)", got, want) + } + first, err := store.FindDirectConversationBySolePeer("google", "karl") + if err != nil || first.ConversationID != "karl-rcs" { + t.Fatalf("FindDirectConversationBySolePeer() = %q, %v; want karl-rcs", first.ConversationID, err) + } + if none, err := store.ListDirectConversationsBySolePeer("google", "shoshana"); err != nil || len(none) != 0 { + t.Fatalf("Shoshana direct candidates = %v, %v; want none", conversationIDs(none), err) + } + if _, err := store.FindDirectConversationBySolePeer("google", "shoshana"); !errors.Is(err, ErrNotFound) { + t.Fatalf("FindDirectConversationBySolePeer(shoshana) err = %v, want ErrNotFound", err) + } + + groups, err := store.ListGroupConversationsByPeerSet("google", []string{"shoshana", "karl"}) + if err != nil { + t.Fatalf("ListGroupConversationsByPeerSet(): %v", err) + } + if got, want := conversationIDs(groups), []string{"book-club", "family"}; !slices.Equal(got, want) { + t.Fatalf("group candidates = %v, want %v (exact rosters only, most recent first)", got, want) + } + if empty, err := store.ListGroupConversationsByPeerSet("google", nil); err != nil || len(empty) != 0 { + t.Fatalf("empty peer set = %v, %v; want none", conversationIDs(empty), err) + } + if _, err := store.FindGroupConversationByPeerSet("google", nil); !errors.Is(err, ErrNotFound) { + t.Fatalf("FindGroupConversationByPeerSet(nil) err = %v, want ErrNotFound", err) + } +} + +func TestRemoteIDSpaceEpochAdvancesOnceFromTheObservedEpoch(t *testing.T) { + store := openIdentityGraphTestStore(t) + seedRebindFixture(t, store.db) + + epoch, err := store.RemoteIDSpaceEpoch("google") + if err != nil || epoch != 0 { + t.Fatalf("initial epoch = %d, %v; want 0", epoch, err) + } + if epoch, err = store.AdvanceRemoteIDSpaceEpoch("google", 0); err != nil || epoch != 1 { + t.Fatalf("advance from 0 = %d, %v; want 1", epoch, err) + } + // A caller still holding epoch 0 must not advance a second time. + if epoch, err = store.AdvanceRemoteIDSpaceEpoch("google", 0); err != nil || epoch != 1 { + t.Fatalf("stale advance from 0 = %d, %v; want 1 unchanged", epoch, err) + } + if epoch, err = store.AdvanceRemoteIDSpaceEpoch("google", 1); err != nil || epoch != 2 { + t.Fatalf("advance from 1 = %d, %v; want 2", epoch, err) + } + if _, err := store.RemoteIDSpaceEpoch("missing"); !errors.Is(err, ErrNotFound) { + t.Fatalf("missing account epoch err = %v, want ErrNotFound", err) + } + + // Account upserts carry no epoch and must leave it alone. + account, err := store.GetAccount("google") + if err != nil { + t.Fatalf("GetAccount(): %v", err) + } + account.DisplayName = "renamed" + if err := store.UpsertAccount(account); err != nil { + t.Fatalf("UpsertAccount(): %v", err) + } + if epoch, err = store.RemoteIDSpaceEpoch("google"); err != nil || epoch != 2 { + t.Fatalf("epoch after account upsert = %d, %v; want 2", epoch, err) + } +} + +func TestConversationRemoteBindingTransitions(t *testing.T) { + store := openIdentityGraphTestStore(t) + seedRebindFixture(t, store.db) + + assertBinding := func(step string, want RemoteBinding) { + t.Helper() + got, err := store.ConversationRemoteBinding("google", "karl-sms") + if err != nil || got != want { + t.Fatalf("%s: binding = %+v (err %v), want %+v", step, got, err, want) + } + } + // Rows written by anything other than ingest's message mint count as + // announced in the starting epoch. + assertBinding("fixture insert", RemoteBinding{AnnouncedEpoch: 0, BoundEpoch: 0}) + + if err := store.MarkConversationProvisional("google", "karl-sms", 1); err != nil { + t.Fatalf("MarkConversationProvisional(): %v", err) + } + assertBinding("message mint", RemoteBinding{Provisional: true, BoundEpoch: 1}) + if err := store.MarkConversationBound("google", "karl-sms", 0); err != nil { + t.Fatalf("MarkConversationBound(0): %v", err) + } + assertBinding("bound never moves backwards", RemoteBinding{Provisional: true, BoundEpoch: 1}) + if err := store.MarkConversationAnnounced("google", "karl-sms", 2); err != nil { + t.Fatalf("MarkConversationAnnounced(2): %v", err) + } + assertBinding("announced", RemoteBinding{AnnouncedEpoch: 2, BoundEpoch: 2}) + if err := store.MarkConversationAnnounced("google", "karl-sms", 1); err != nil { + t.Fatalf("MarkConversationAnnounced(1): %v", err) + } + assertBinding("announcement never moves backwards", RemoteBinding{AnnouncedEpoch: 2, BoundEpoch: 2}) + + conversation, err := store.GetConversation("karl-sms") + if err != nil { + t.Fatalf("GetConversation(): %v", err) + } + conversation.Title = "Karl (SMS)" + if err := store.UpsertConversation(conversation); err != nil { + t.Fatalf("UpsertConversation(): %v", err) + } + assertBinding("after a conversation upsert", RemoteBinding{AnnouncedEpoch: 2, BoundEpoch: 2}) + if err := store.ReassignConversationRemoteID("google", "7001", "karl-sms", rebindTestTimeMS); err != nil { + t.Fatalf("ReassignConversationRemoteID(): %v", err) + } + assertBinding("after a store-level rebind", RemoteBinding{AnnouncedEpoch: 2, BoundEpoch: 2}) + + for name, mark := range map[string]func(string, string, int64) error{ + "announced": store.MarkConversationAnnounced, + "provisional": store.MarkConversationProvisional, + "bound": store.MarkConversationBound, + } { + if err := mark("google", "missing", 0); !errors.Is(err, ErrNotFound) { + t.Fatalf("mark %s on a missing conversation: err = %v, want ErrNotFound", name, err) + } + if err := mark("google", "karl-sms", -1); err == nil { + t.Fatalf("mark %s accepted a negative epoch", name) + } + } + if _, err := store.ConversationRemoteBinding("google", "missing"); !errors.Is(err, ErrNotFound) { + t.Fatalf("missing conversation binding err = %v, want ErrNotFound", err) + } +} + +func TestMergeConversationIntoMovesHistoryAndBindsTheWireID(t *testing.T) { + store := openIdentityGraphTestStore(t) + seedRebindFixture(t, store.db) + // karl-rcs (6002) was minted for a thread that is really karl-sms's + // continuation: it holds a re-served copy of a karl-sms message, one new + // message, and one whose remote ID karl-sms already uses for another message. + mustExec(t, store.db, ` + INSERT INTO messages (message_id, conversation_id, account_id, remote_message_id, sender_identity_id, direction, body, occurred_at_ms, created_at_ms, updated_at_ms) + VALUES + ('old-1', 'karl-sms', 'google', '100', 'karl', 'incoming', 'hello', 1000, ?1, ?1), + ('old-2', 'karl-sms', 'google', '300', 'karl', 'incoming', 'older 300', 900, ?1, ?1), + ('dup-1', 'karl-rcs', 'google', '200', 'karl', 'incoming', 'hello', 1000, ?1, ?1), + ('new-1', 'karl-rcs', 'google', '201', 'karl', 'incoming', 'new', 5000, ?1, ?1), + ('clash', 'karl-rcs', 'google', '300', 'karl', 'incoming', 'newer 300', 6000, ?1, ?1) + `, rebindTestTimeMS) + + merge, err := store.MergeConversationInto(context.Background(), "google", "6002", "karl-rcs", "karl-sms", rebindTestTimeMS) + if err != nil { + t.Fatalf("MergeConversationInto(): %v", err) + } + if merge != (ConversationMerge{Moved: 1, Duplicates: 1, Kept: 1}) { + t.Fatalf("merge = %+v, want 1 moved, 1 duplicate, 1 kept", merge) + } + target, err := store.GetConversationByRemote("google", "6002") + if err != nil || target.ConversationID != "karl-sms" { + t.Fatalf("6002 bound to %q (err %v), want karl-sms", target.ConversationID, err) + } + if target.LastMessageAtMS != 5000 { + t.Fatalf("karl-sms recency = %d, want 5000 from the moved message", target.LastMessageAtMS) + } + source, err := store.GetConversation("karl-rcs") + if err != nil { + t.Fatalf("source with a kept message was dropped: %v", err) + } + if source.RemoteConversationID != displacedRemoteID("6002", "karl-rcs") { + t.Fatalf("source remote ID = %q, want displaced", source.RemoteConversationID) + } + var homes = map[string]string{} + rows, err := store.db.Query(`SELECT message_id, conversation_id FROM messages`) + if err != nil { + t.Fatalf("list messages: %v", err) + } + for rows.Next() { + var id, conversation string + if err := rows.Scan(&id, &conversation); err != nil { + t.Fatalf("scan: %v", err) + } + homes[id] = conversation + } + rows.Close() + want := map[string]string{"old-1": "karl-sms", "old-2": "karl-sms", "new-1": "karl-sms", "clash": "karl-rcs"} + if len(homes) != len(want) { + t.Fatalf("messages after merge = %v, want %v (the duplicate deleted)", homes, want) + } + for id, conversation := range want { + if homes[id] != conversation { + t.Fatalf("message %s in %q, want %q", id, homes[id], conversation) + } + } + + // A source left empty is deleted; one that no longer holds the wire ID is + // refused. + mustExec(t, store.db, ` + INSERT INTO conversations (conversation_id, account_id, remote_conversation_id, kind, created_at_ms, updated_at_ms) + VALUES ('empty', 'google', '6003', 'direct', ?1, ?1) + `, rebindTestTimeMS) + merge, err = store.MergeConversationInto(context.Background(), "google", "6003", "empty", "karl-sms", rebindTestTimeMS) + if err != nil || !merge.Dropped { + t.Fatalf("merge of an empty source = %+v, %v; want dropped", merge, err) + } + if _, err := store.GetConversation("empty"); !errors.Is(err, ErrNotFound) { + t.Fatalf("empty source still present: %v", err) + } + if _, err := store.MergeConversationInto(context.Background(), "google", "6001", "karl-rcs", "karl-sms", rebindTestTimeMS); !errors.Is(err, ErrNotFound) { + t.Fatalf("merge of a source that does not hold the wire ID: err = %v, want ErrNotFound", err) + } +} + +// Migration 0011 leaves every existing row announced in epoch 0 with its +// binding in epoch 0, so no existing thread becomes re-keyable by a +// same-roster twin on upgrade. +func TestRemoteIDSpaceMigrationDefaultsExistingV10Rows(t *testing.T) { + path := filepath.Join(t.TempDir(), "store.sqlite3") + database, err := sql.Open("sqlite", storeDSN(path)) + if err != nil { + t.Fatalf("sql.Open(): %v", err) + } + ctx := context.Background() + if err := enableWAL(ctx, database); err != nil { + _ = database.Close() + t.Fatalf("enableWAL(): %v", err) + } + if err := verifyConnectionPragmas(ctx, database); err != nil { + _ = database.Close() + t.Fatalf("verifyConnectionPragmas(): %v", err) + } + if err := runMigrations(ctx, database, embeddedMigrations[:10]); err != nil { + _ = database.Close() + t.Fatalf("runMigrations(v10): %v", err) + } + seedRebindFixture(t, database) + before := readLedgerRows(t, database) + if err := database.Close(); err != nil { + t.Fatalf("close v10 database: %v", err) + } + + store, err := Open(path) + if err != nil { + t.Fatalf("Open(v10): %v", err) + } + t.Cleanup(func() { + if err := store.Close(); err != nil { + t.Errorf("Close(): %v", err) + } + }) + after := readLedgerRows(t, store.db) + if len(after) != 11 || !slices.Equal(after[:10], before) { + t.Fatalf("ledger after upgrade = %+v, want 0001-0010 unchanged plus 0011", after) + } + if after[10].name != "remote_idspace_epochs" { + t.Fatalf("migration 0011 name = %q", after[10].name) + } + assertPragmaInt(t, store.db, "user_version", 11) + + if epoch, err := store.RemoteIDSpaceEpoch("google"); err != nil || epoch != 0 { + t.Fatalf("account epoch after upgrade = %d, %v; want 0", epoch, err) + } + for _, id := range []string{"karl-sms", "karl-rcs", "karl-old", "family", "book-club", "trio"} { + binding, err := store.ConversationRemoteBinding("google", id) + if err != nil || binding != (RemoteBinding{}) { + t.Fatalf("%s after upgrade: %+v (err %v), want announced and bound in epoch 0", id, binding, err) + } + } + expectExecError(t, store.db, "negative account epoch", + `UPDATE accounts SET remote_idspace_epoch = -1 WHERE account_id = 'google'`) + expectExecError(t, store.db, "negative announced epoch", + `UPDATE conversations SET remote_announced_epoch = -1 WHERE conversation_id = 'family'`) + expectExecError(t, store.db, "negative bound epoch", + `UPDATE conversations SET remote_bound_epoch = -1 WHERE conversation_id = 'family'`) +}