From d52537440e24c5d72f7d89a8a1d5fa1d068e4218 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Thu, 8 Oct 2026 18:19:18 -0400 Subject: [PATCH] Keep same-roster Google threads distinct; re-key only threads whose id is dead Two groups with the same members, or two 1:1 threads with one person, are both live on the phone under their own wire ids. The #176 roster re-key moved one's id onto the other's row, so the id and title flipped on every event and both threads' messages piled into one row. A thread now takes a new wire id only when its own id is dead: displaced by a collision, or announced by a ConversationEvent only in an earlier device ID space. - Migration 0011 adds accounts.remote_idspace_epoch and, per conversation, remote_announced_epoch (NULL = provisional, minted from a message) and remote_bound_epoch. Existing and externally written rows default to epoch 0. - A new epoch starts when an id the phone announced in the current epoch arrives naming another thread. Declines made in the 15 minutes before that detection are merged into the threads they continue. - Group re-keys also require the title to match. - A provisional thread that hears from a second sender becomes a group instead of being scattered across its members' 1:1 threads. - Google reuses message ids across a reset; a continued thread's older message keeps its row under a displaced remote id instead of being overwritten. - A ConversationEvent after a message whose phone timestamp leads this clock no longer fails updated_at_ms >= created_at_ms and gets quarantined. Co-Authored-By: Claude Opus 5.5 --- docs/agent-runbook.md | 21 + internal/ingest/counters.go | 27 + internal/ingest/idspace.go | 591 +++++++++++++++++- .../ingest/idspace_distinct_threads_test.go | 581 +++++++++++++++++ internal/ingest/idspace_properties_test.go | 341 ++++++++++ internal/ingest/worker.go | 37 +- internal/migration/transform_test.go | 4 +- internal/migration/validate.go | 8 +- internal/storage/sqlite/attachments_test.go | 4 +- .../storage/sqlite/identity_graph_test.go | 4 +- .../sqlite/message_attachments_test.go | 10 +- internal/storage/sqlite/messages.go | 33 + internal/storage/sqlite/messages_test.go | 4 +- internal/storage/sqlite/migrations.go | 4 + .../migrations/0011_remote_idspace_epochs.sql | 26 + internal/storage/sqlite/outbox_test.go | 20 +- .../sqlite/reactions_migration_test.go | 10 +- internal/storage/sqlite/rebind.go | 382 ++++++++++- internal/storage/sqlite/rebind_test.go | 337 ++++++++++ 19 files changed, 2366 insertions(+), 78 deletions(-) create mode 100644 internal/ingest/idspace_distinct_threads_test.go create mode 100644 internal/ingest/idspace_properties_test.go create mode 100644 internal/storage/sqlite/migrations/0011_remote_idspace_epochs.sql create mode 100644 internal/storage/sqlite/rebind_test.go 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'`) +}