diff --git a/docs/agent-runbook.md b/docs/agent-runbook.md index 371f0343..76ac8e7c 100644 --- a/docs/agent-runbook.md +++ b/docs/agent-runbook.md @@ -93,7 +93,7 @@ fix (this PR) makes v2 reads accept **either** key: unknown ids fall back to a carries the canonical v2 id. Prefer storing the **v2 id** for anything durable; the alias exists so old references keep working. -Three more cutover artifacts worth knowing: +Four more cutover artifacts worth knowing: - **Only the Google decoder emits `ConversationEvent` frames.** Signal and WhatsApp conversations are minted from message frames, which carry no kind and @@ -130,15 +130,47 @@ curl -s http://127.0.0.1:7007/api/status | jq '.freshness' `UNIQUE constraint failed: devices.account_id`. Each call first rewrote the account's `bridge_key` (`google_messages` → `google`), and v2 reads derive a thread's platform from that key, so Google threads would then read as - `google` instead of `sms`. A v2-primary daemon no longer runs the mirror on - `/api/mark-read`. The keys should read `google_messages`, `whatsmeow` and - `signal_cli` (the live store was clean on 2026-10-09): + `google` instead of `sms`. A v2-primary daemon does not run the mirror on + `/api/mark-read`; it writes the cursor natively (next bullet). The keys + should read `google_messages`, `whatsmeow` and `signal_cli` (the live store + was clean on 2026-10-09): ```bash sqlite3 -readonly "$HOME/Library/Application Support/OpenMessage/v2/store.sqlite3" \ "SELECT account_id, bridge_key, display_name FROM accounts" ``` +- **On v2-primary, `/api/mark-read` writes the v2 read cursor natively** + (`v2wire.MarkReadV2`). It resolves `conversation_id` the way v2 reads do: + the v2 id, or a legacy-form id through `remote_conversation_id` + (`v2read.ResolveConversation`). An id that matches nothing writes nothing; + the mirror would have created the thread under its legacy id. The cursor + belongs to the account's local installation device, the one + `GetLocalInstallationDevice` resolves and ingest receipts advance, so on a + migrated store it is the derived-id device. An account with no local device + gets one under the migration's derived id + (`v2keys.LocalInstallationDeviceID`), never `local-primary:`. The + cursor is dated at the request and names no message + (`last_read_message_id` NULL, as the mirror writes it). `read_cursors` + references `messages` with no `ON DELETE` action, so a cursor that named the + newest message would pin it. The outbox deletes a send's echo duplicate on + reconcile, and that delete then fails with `FOREIGN KEY constraint failed`. + Id-space repair likewise leaves cursor-referenced messages in place. Writes + are monotone in read time, so a cursor already dated later than the request + stays (an ingested receipt carries the receipt's own time). The write is best + effort and sends no read receipt to the phone. A failure logs `Failed to write v2 read cursor` with + `conv_id`, and the response stays 200. Nothing reads these cursors yet: v2 + reads report `UnreadCount: 0`, and the web UI calls mark-read only when + `UnreadCount > 0`, so on v2-primary only API callers reach it until unread + derivation (S5b/S8) lands. To see the latest cursors: + +```bash +sqlite3 -readonly "$HOME/Library/Application Support/OpenMessage/v2/store.sqlite3" \ + "SELECT conversation_id, device_id, last_read_message_id, + datetime(last_read_at_ms / 1000, 'unixepoch', 'localtime') + FROM read_cursors ORDER BY last_read_at_ms DESC LIMIT 5" +``` + ## Google thread ids are device-local: phone swaps re-key everything Google Messages conversation and message ids are the phone's own row ids, not diff --git a/internal/migration/transform.go b/internal/migration/transform.go index 787612ad..b17eb6a8 100644 --- a/internal/migration/transform.go +++ b/internal/migration/transform.go @@ -615,7 +615,7 @@ func writeAccounts(target *sqlite.Store, state *transformState) error { }); err != nil { return fmt.Errorf("upsert %s account: %w", platform, err) } - deviceID := v2keys.DeriveID("device", account.AccountID, account.AccountID+"\x1flocal") + deviceID := v2keys.LocalInstallationDeviceID(account.AccountID) if err := target.UpsertDevice(sqlite.Device{ DeviceID: deviceID, AccountID: account.AccountID, Kind: sqlite.DeviceKindLocalInstallation, DisplayName: "OpenMessage", @@ -984,10 +984,7 @@ func writeReadCursors( lastReadID = &value lastReadAt = positions[index].At } - deviceID := v2keys.DeriveID( - "device", conversation.Account.AccountID, - conversation.Account.AccountID+"\x1flocal", - ) + deviceID := v2keys.LocalInstallationDeviceID(conversation.Account.AccountID) updatedAt := state.baseTimestampMS if len(positions) > 0 { updatedAt = maxInt64(updatedAt, positions[len(positions)-1].At) diff --git a/internal/v2keys/derive.go b/internal/v2keys/derive.go index 2bbc620d..fee7901e 100644 --- a/internal/v2keys/derive.go +++ b/internal/v2keys/derive.go @@ -26,6 +26,15 @@ func DeriveID(entity, accountID, naturalKey string) string { return hex.EncodeToString(sum[:])[:32] } +// LocalInstallationDeviceID is the ID the migration mints for an account's +// local installation device. Code that must create that device for an account +// that has none uses it too, so a v2 store holds one ID shape whichever path +// created the device. Code that reads the device resolves it by role +// (sqlite.Store.GetLocalInstallationDevice), never by this ID. +func LocalInstallationDeviceID(accountID string) string { + return DeriveID("device", accountID, accountID+"\x1flocal") +} + var signalACI = regexp.MustCompile(`(?i)^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$`) // IdentityKey classifies and canonicalizes a platform identity. diff --git a/internal/v2keys/derive_test.go b/internal/v2keys/derive_test.go index 8bd5f69b..b40f399f 100644 --- a/internal/v2keys/derive_test.go +++ b/internal/v2keys/derive_test.go @@ -44,6 +44,23 @@ func TestDeriveIDMigrationFixtureGoldens(t *testing.T) { } } +// TestLocalInstallationDeviceIDMatchesMigrationGoldens pins the helper to the +// device IDs the migration minted before the helper existed, so a device the +// native mark-read path creates has the ID a migrated store would hold. +func TestLocalInstallationDeviceIDMatchesMigrationGoldens(t *testing.T) { + t.Parallel() + + for accountID, want := range map[string]string{ + "google-primary": "9bcc134365b6f21de496ec1693b68421", + "whatsapp-primary": "1aa97228aad2f8a40a7eea579625d579", + "signal-primary": "f5725a0b516450efeab65b0ccdc9d041", + } { + if got := LocalInstallationDeviceID(accountID); got != want { + t.Errorf("LocalInstallationDeviceID(%q) = %q, want migration golden %q", accountID, got, want) + } + } +} + func TestIdentityKey(t *testing.T) { t.Parallel() diff --git a/internal/v2read/alias.go b/internal/v2read/alias.go index 64275a5d..d5ca8450 100644 --- a/internal/v2read/alias.go +++ b/internal/v2read/alias.go @@ -2,6 +2,7 @@ package v2read import ( "errors" + "fmt" "strings" "github.com/maxghenis/openmessage/internal/storage/sqlite" @@ -9,29 +10,49 @@ import ( ) // resolveConversationID maps a caller-supplied conversation key onto the v2 -// primary key. v2 conversation IDs pass through untouched. Any other value is -// tried as a remote conversation ID — the key the legacy store and every -// pre-cutover consumer used ("signal:+15551234567", "signal-group:…", a -// Google Messages thread id, a WhatsApp JID) — across all accounts. Cutover -// re-keys every conversation to a derived hash, so without this fallback each -// stored legacy ID silently reads as an empty conversation the moment a -// restart flips the app to v2-primary (issue #155). -// -// Unknown keys return unchanged so callers keep their existing +// primary key with ResolveConversation. Unknown keys, and keys whose lookup +// fails, return unchanged so callers keep their existing // empty-result/not-found semantics. func (s *Source) resolveConversationID(id string) string { trimmed := strings.TrimSpace(id) if trimmed == "" { return id } - if _, err := s.store.GetConversation(trimmed); err == nil { - return trimmed - } else if !errors.Is(err, sqlite.ErrNotFound) { + conversation, err := ResolveConversation(s.store, trimmed) + if err != nil { return trimmed } - accounts, err := s.store.ListAccounts() + return conversation.ConversationID +} + +// ResolveConversation returns the v2 conversation a caller-supplied key names. +// A v2 conversation ID resolves to itself. Any other value is tried as a +// remote conversation ID — the key the legacy store and every pre-cutover +// consumer used ("signal:+15551234567", "signal-group:…", a Google Messages +// thread id, a WhatsApp JID) — across all accounts. Cutover re-keys every +// conversation to a derived hash, so without this fallback each stored legacy +// ID silently reads as an empty conversation the moment a restart flips the +// app to v2-primary (issue #155). +// +// It is the one alias rule for v2 reads and for writes that take a +// caller-supplied conversation ID (v2wire.MarkReadV2), so a stored legacy ID +// writes to the same thread it reads. A key that matches nothing returns an +// error wrapping sqlite.ErrNotFound. +func ResolveConversation(store *sqlite.Store, id string) (sqlite.Conversation, error) { + trimmed := strings.TrimSpace(id) + if trimmed == "" { + return sqlite.Conversation{}, fmt.Errorf("resolve conversation: empty id: %w", sqlite.ErrNotFound) + } + conversation, err := store.GetConversation(trimmed) + if err == nil { + return conversation, nil + } + if !errors.Is(err, sqlite.ErrNotFound) { + return sqlite.Conversation{}, fmt.Errorf("resolve conversation %q: %w", trimmed, err) + } + accounts, err := store.ListAccounts() if err != nil { - return trimmed + return sqlite.Conversation{}, fmt.Errorf("resolve conversation %q: %w", trimmed, err) } var best *sqlite.Conversation for _, account := range accounts { @@ -39,7 +60,7 @@ func (s *Source) resolveConversationID(id string) string { platformForBridgeKey(account.BridgeKey), trimmed, ) - conversation, err := s.store.GetConversationByRemote(account.AccountID, remoteID) + conversation, err := store.GetConversationByRemote(account.AccountID, remoteID) if err != nil { continue } @@ -56,7 +77,7 @@ func (s *Source) resolveConversationID(id string) string { } } if best != nil { - return best.ConversationID + return *best, nil } - return trimmed + return sqlite.Conversation{}, fmt.Errorf("resolve conversation %q: %w", trimmed, sqlite.ErrNotFound) } diff --git a/internal/v2read/alias_resolve_test.go b/internal/v2read/alias_resolve_test.go new file mode 100644 index 00000000..4baa9e8f --- /dev/null +++ b/internal/v2read/alias_resolve_test.go @@ -0,0 +1,167 @@ +package v2read + +import ( + "database/sql" + "errors" + "math/rand" + "strings" + "testing" + "testing/quick" + + "github.com/maxghenis/openmessage/internal/storage/sqlite" + "github.com/maxghenis/openmessage/internal/v2keys" +) + +func TestResolveConversationReturnsCanonicalRowOrNotFound(t *testing.T) { + store, _, _ := openSourceTestStore(t) + seedSourceAccount(t, store, "signal-primary", "signal_cli") + const remoteID = "signal:+15550000002" + conversationID := v2keys.DeriveID("conversation", "signal-primary", remoteID) + seedSourceConversation(t, store, sqlite.Conversation{ + ConversationID: conversationID, + AccountID: "signal-primary", + RemoteConversationID: remoteID, + Kind: sqlite.ConversationKindDirect, + NotificationMode: sqlite.NotificationModeAll, + LastMessageAtMS: 400, + }) + + for _, key := range []string{ + conversationID, + " " + conversationID + "\n", + remoteID, + "\t" + remoteID + " ", + // Signal remote IDs normalize the address payload as they did at + // cutover. + "signal: +15550000002", + } { + conversation, err := ResolveConversation(store, key) + if err != nil { + t.Fatalf("ResolveConversation(%q): %v", key, err) + } + if conversation.ConversationID != conversationID || conversation.AccountID != "signal-primary" { + t.Fatalf("ResolveConversation(%q) = %+v, want %q", key, conversation, conversationID) + } + } + for _, key := range []string{"", " ", "signal:+15550009999", v2keys.DeriveID("conversation", "signal-primary", "absent")} { + if _, err := ResolveConversation(store, key); !errors.Is(err, sqlite.ErrNotFound) { + t.Fatalf("ResolveConversation(%q) error = %v, want ErrNotFound", key, err) + } + } +} + +// TestResolveConversationMatchesReferenceAndReadPath checks, over random +// stores whose remote IDs collide across accounts and whose recency ties, +// that ResolveConversation agrees with a brute-force reading of the alias +// rule, and that the v2 read path (Source.GetConversation) serves the same +// conversation. The second check is what lets a write keyed by a +// caller-supplied ID land on the thread a read of that ID shows. +func TestResolveConversationMatchesReferenceAndReadPath(t *testing.T) { + accounts := []struct { + accountID string + bridgeKey string + }{ + {"google-primary", "google_messages"}, + {"signal-primary", "signal_cli"}, + {"whatsapp-primary", "whatsmeow"}, + } + remotePool := []string{ + "thread-1", + "thread-2", + "signal:+15550000001", + "signal-group:QUJD=", + "15550000003@s.whatsapp.net", + } + property := func(seed int64) bool { + random := rand.New(rand.NewSource(seed)) + store, _, source := openSourceTestStore(t) + var conversations []sqlite.Conversation + platformOf := map[string]string{} + for _, account := range accounts { + seedSourceAccount(t, store, account.accountID, account.bridgeKey) + platformOf[account.accountID] = platformForBridgeKey(account.bridgeKey) + for _, remoteID := range remotePool { + if random.Intn(2) == 0 { + continue + } + conversation := sqlite.Conversation{ + ConversationID: v2keys.DeriveID("conversation", account.accountID, remoteID), + AccountID: account.accountID, + RemoteConversationID: remoteID, + Kind: sqlite.ConversationKindDirect, + NotificationMode: sqlite.NotificationModeAll, + // Few distinct values, so recency ties are common. + LastMessageAtMS: int64(random.Intn(3)) * 100, + } + seedSourceConversation(t, store, conversation) + conversations = append(conversations, conversation) + } + } + + // reference is the alias rule written out over the seeded rows. + reference := func(key string) (string, bool) { + trimmed := strings.TrimSpace(key) + if trimmed == "" { + return "", false + } + for _, conversation := range conversations { + if conversation.ConversationID == trimmed { + return trimmed, true + } + } + var best *sqlite.Conversation + for index := range conversations { + conversation := &conversations[index] + if conversation.RemoteConversationID != v2keys.NormalizeRemoteConversationID(platformOf[conversation.AccountID], trimmed) { + continue + } + if best == nil || conversation.LastMessageAtMS > best.LastMessageAtMS || + (conversation.LastMessageAtMS == best.LastMessageAtMS && conversation.ConversationID < best.ConversationID) { + best = conversation + } + } + if best == nil { + return "", false + } + return best.ConversationID, true + } + + keys := []string{"", " ", "unknown-thread", "signal: +15550000001", "signal-group: QUJD="} + for _, remoteID := range remotePool { + keys = append(keys, remoteID, " "+remoteID+"\t") + } + for _, account := range accounts { + for _, remoteID := range remotePool { + keys = append(keys, v2keys.DeriveID("conversation", account.accountID, remoteID)) + } + } + for _, key := range keys { + wantID, wantFound := reference(key) + resolved, err := ResolveConversation(store, key) + if wantFound != (err == nil) || (err != nil && !errors.Is(err, sqlite.ErrNotFound)) { + t.Errorf("seed %d: ResolveConversation(%q) error = %v, want found=%v", seed, key, err, wantFound) + return false + } + if wantFound && resolved.ConversationID != wantID { + t.Errorf("seed %d: ResolveConversation(%q) = %q, reference %q", seed, key, resolved.ConversationID, wantID) + return false + } + read, err := source.GetConversation(key) + if !wantFound { + if !errors.Is(err, sql.ErrNoRows) { + t.Errorf("seed %d: GetConversation(%q) = %+v, %v; want sql.ErrNoRows", seed, key, read, err) + return false + } + continue + } + if err != nil || read.ConversationID != wantID { + t.Errorf("seed %d: GetConversation(%q) = %+v, %v; want %q", seed, key, read, err, wantID) + return false + } + } + return true + } + if err := quick.Check(property, &quick.Config{MaxCount: 40, Rand: rand.New(rand.NewSource(20261010))}); err != nil { + t.Fatal(err) + } +} diff --git a/internal/v2wire/read_native.go b/internal/v2wire/read_native.go new file mode 100644 index 00000000..20a4cec1 --- /dev/null +++ b/internal/v2wire/read_native.go @@ -0,0 +1,104 @@ +package v2wire + +import ( + "context" + "errors" + "fmt" + + "github.com/maxghenis/openmessage/internal/storage/sqlite" + "github.com/maxghenis/openmessage/internal/v2keys" + "github.com/maxghenis/openmessage/internal/v2read" +) + +// MarkReadV2 records a v2-primary mark-read as the read cursor of the account's +// local installation device. It writes the v2 store only and sends no +// transport read receipt, like MirrorReadCursor on a legacy-primary daemon. +// +// conversationID is the v2 ID the UI sends, or a legacy-form ID resolved with +// the alias fallback v2 reads use (v2read.ResolveConversation), so a stored +// legacy ID marks the thread it reads. An ID that resolves to no conversation +// writes nothing and returns an error wrapping sqlite.ErrNotFound: unlike the +// legacy mirror, this path never creates a conversation, account, or row +// keyed by a legacy ID. +// +// The cursor lands on the device GetLocalInstallationDevice resolves, the one +// ingest advances receipt cursors for, whatever its ID (a migrated store keys +// it by hash). An account with no local installation device gets one under +// v2keys.LocalInstallationDeviceID, the ID the migration would have minted. +// +// The cursor names no message (LastReadMessageID nil), as in the legacy +// mirror: it says the thread was read as of atMS. read_cursors references +// messages without an ON DELETE action, so a cursor that named the newest +// message would pin it. The newest message is the one most likely to be a +// send's echo duplicate, which the outbox deletes on reconcile, and that +// delete would then fail. UpsertReadCursor is monotone in read time, so an +// atMS older than the stored cursor leaves it unchanged. +func MarkReadV2( + ctx context.Context, + v2 *sqlite.Store, + conversationID string, + atMS int64, +) error { + if ctx == nil { + return errors.New("mark read v2: context is nil") + } + if err := ctx.Err(); err != nil { + return fmt.Errorf("mark read v2: %w", err) + } + if v2 == nil { + return errors.New("mark read v2: v2 store is nil") + } + if atMS <= 0 { + return errors.New("mark read v2: timestamp must be positive") + } + conversation, err := v2read.ResolveConversation(v2, conversationID) + if err != nil { + return fmt.Errorf("mark read v2: %w", err) + } + device, err := localInstallationDevice(ctx, v2, conversation.AccountID, atMS) + if err != nil { + return fmt.Errorf("mark read v2 conversation %q: %w", conversation.ConversationID, err) + } + if err := v2.UpsertReadCursor(sqlite.ReadCursor{ + AccountID: conversation.AccountID, + DeviceID: device.DeviceID, + ConversationID: conversation.ConversationID, + LastReadMessageID: nil, + LastReadAtMS: atMS, + UpdatedAtMS: atMS, + }); err != nil { + return fmt.Errorf("mark read v2 conversation %q: %w", conversation.ConversationID, err) + } + return nil +} + +// localInstallationDevice returns the account's local installation device, +// creating it under the migration's derived ID only when the account has none. +func localInstallationDevice( + ctx context.Context, + v2 *sqlite.Store, + accountID string, + nowMS int64, +) (sqlite.Device, error) { + device, err := v2.GetLocalInstallationDevice(ctx, accountID) + if err == nil { + return device, nil + } + if !errors.Is(err, sqlite.ErrNotFound) { + return sqlite.Device{}, err + } + device, err = v2.EnsureLocalInstallationDevice(ctx, sqlite.Device{ + DeviceID: v2keys.LocalInstallationDeviceID(accountID), + AccountID: accountID, + Kind: sqlite.DeviceKindLocalInstallation, + DisplayName: "OpenMessage", + State: sqlite.DeviceStateActive, + IsCurrent: true, + CreatedAtMS: nowMS, + UpdatedAtMS: nowMS, + }) + if err != nil { + return sqlite.Device{}, fmt.Errorf("ensure local device for account %q: %w", accountID, err) + } + return device, nil +} diff --git a/internal/v2wire/read_native_test.go b/internal/v2wire/read_native_test.go new file mode 100644 index 00000000..713a7bbb --- /dev/null +++ b/internal/v2wire/read_native_test.go @@ -0,0 +1,459 @@ +package v2wire + +import ( + "context" + "errors" + "fmt" + "math/rand" + "reflect" + "slices" + "testing" + "testing/quick" + + "github.com/maxghenis/openmessage/internal/storage/sqlite" + "github.com/maxghenis/openmessage/internal/v2keys" +) + +// TestMarkReadV2OnMigratedStoreWritesDerivedDeviceCursor is the v2-primary +// shape of the live install: the migration keyed every conversation and each +// account's local device by hash. Marking a thread read by its v2 id or by +// the legacy id it had before cutover writes the same cursor, on the derived +// device, and touches nothing else. +func TestMarkReadV2OnMigratedStoreWritesDerivedDeviceCursor(t *testing.T) { + _, v2 := openMigratedTestStores(t) + ctx := context.Background() + for _, conversation := range migratedFixtureConversations { + t.Run(conversation.legacyID, func(t *testing.T) { + owner := mustMigratedConversation(t, v2, conversation) + deviceID := v2keys.LocalInstallationDeviceID(conversation.accountID) + migratedCursor, err := v2.GetReadCursor(deviceID, owner.ConversationID) + if err != nil { + t.Fatalf("migrated GetReadCursor(): %v", err) + } + if migratedCursor.LastReadMessageID == nil { + t.Fatalf("migrated cursor = %+v; fixture no longer starts from a positioned cursor", migratedCursor) + } + before := snapshotV2(t, v2) + + readAtMS := int64(1_910_000_000_000) + for _, key := range []string{owner.ConversationID, conversation.legacyID, " " + conversation.legacyID + "\t"} { + readAtMS += 1_000 + if err := MarkReadV2(ctx, v2, key, readAtMS); err != nil { + t.Fatalf("MarkReadV2(%q): %v", key, err) + } + cursor, err := v2.GetReadCursor(deviceID, owner.ConversationID) + if err != nil { + t.Fatalf("GetReadCursor(derived device) after %q: %v", key, err) + } + want := sqlite.ReadCursor{ + AccountID: conversation.accountID, DeviceID: deviceID, ConversationID: owner.ConversationID, + LastReadAtMS: readAtMS, UpdatedAtMS: readAtMS, + } + if !reflect.DeepEqual(cursor, want) { + t.Fatalf("cursor after %q = %+v, want %+v", key, cursor, want) + } + } + + after := snapshotV2(t, v2) + delete(after.cursors, cursorKey{deviceID, owner.ConversationID}) + delete(before.cursors, cursorKey{deviceID, owner.ConversationID}) + if !reflect.DeepEqual(after, before) { + t.Fatalf("v2 rows besides the cursor changed:\nbefore %+v\nafter %+v", before, after) + } + if _, err := v2.GetDevice(localDeviceID(conversation.accountID)); !errors.Is(err, sqlite.ErrNotFound) { + t.Fatalf("GetDevice(%q) error = %v, want ErrNotFound", localDeviceID(conversation.accountID), err) + } + }) + } +} + +// TestMarkReadV2UnknownConversationWritesNothing covers ids the v2 store does +// not know, including legacy ids of threads that only reached the legacy +// store after cutover. The legacy mirror would create those threads; the +// native path refuses without writing. +func TestMarkReadV2UnknownConversationWritesNothing(t *testing.T) { + _, v2 := openMigratedTestStores(t) + ctx := context.Background() + before := snapshotV2(t, v2) + keys := []string{"", " ", "not-a-conversation", v2keys.DeriveID("conversation", googleAccountID, "nowhere")} + for _, conversation := range postCutoverFixtureConversations { + keys = append(keys, conversation.legacyID) + } + for _, key := range keys { + if err := MarkReadV2(ctx, v2, key, 1_910_000_000_000); !errors.Is(err, sqlite.ErrNotFound) { + t.Fatalf("MarkReadV2(%q) error = %v, want ErrNotFound", key, err) + } + } + if after := snapshotV2(t, v2); !reflect.DeepEqual(after, before) { + t.Fatalf("v2 rows changed:\nbefore %+v\nafter %+v", before, after) + } +} + +func TestMarkReadV2RejectsInvalidInputWithoutWriting(t *testing.T) { + _, v2 := openMigratedTestStores(t) + owner := mustMigratedConversation(t, v2, migratedFixtureConversations[0]) + canceled, cancel := context.WithCancel(context.Background()) + cancel() + before := snapshotV2(t, v2) + for name, call := range map[string]func() error{ + "nil context": func() error { return MarkReadV2(nil, v2, owner.ConversationID, 1_910_000_000_000) }, + "canceled context": func() error { return MarkReadV2(canceled, v2, owner.ConversationID, 1_910_000_000_000) }, + "nil store": func() error { return MarkReadV2(context.Background(), nil, owner.ConversationID, 1_910_000_000_000) }, + "zero timestamp": func() error { return MarkReadV2(context.Background(), v2, owner.ConversationID, 0) }, + "negative timestamp": func() error { return MarkReadV2(context.Background(), v2, owner.ConversationID, -1) }, + } { + if err := call(); err == nil { + t.Errorf("%s: MarkReadV2() error = nil, want an error", name) + } + } + if after := snapshotV2(t, v2); !reflect.DeepEqual(after, before) { + t.Fatalf("v2 rows changed:\nbefore %+v\nafter %+v", before, after) + } +} + +// TestMarkReadV2CreatesDerivedLocalDeviceOnlyWhenAccountHasNone covers an +// account the live bootstrap created after cutover, which has no local +// installation device, next to one whose only local device is non-current and +// keyed by neither the derived nor the mirror ID. +func TestMarkReadV2CreatesDerivedLocalDeviceOnlyWhenAccountHasNone(t *testing.T) { + v2 := openV2TestStore(t) + ctx := context.Background() + seedMarkReadAccount(t, v2, googleAccountID) + seedMarkReadAccount(t, v2, signalAccountID) + google := seedMarkReadConversation(t, v2, googleAccountID, "google-thread") + signal := seedMarkReadConversation(t, v2, signalAccountID, "signal:+15550004444") + const otherDeviceID = "pre-existing-local" + if err := v2.UpsertDevice(sqlite.Device{ + DeviceID: otherDeviceID, AccountID: signalAccountID, Kind: sqlite.DeviceKindLocalInstallation, + DisplayName: "OpenMessage", State: sqlite.DeviceStateActive, IsCurrent: false, + CreatedAtMS: 1_800_000_000_000, UpdatedAtMS: 1_800_000_000_000, + }); err != nil { + t.Fatalf("UpsertDevice(): %v", err) + } + signalDevicesBefore := mustListDevices(t, v2, signalAccountID) + + for _, readAtMS := range []int64{1_910_000_000_000, 1_910_000_000_500} { + if err := MarkReadV2(ctx, v2, "google-thread", readAtMS); err != nil { + t.Fatalf("MarkReadV2(google): %v", err) + } + if err := MarkReadV2(ctx, v2, "signal:+15550004444", readAtMS); err != nil { + t.Fatalf("MarkReadV2(signal): %v", err) + } + } + + devices := mustListDevices(t, v2, googleAccountID) + derived := v2keys.LocalInstallationDeviceID(googleAccountID) + if len(devices) != 1 || devices[0].DeviceID != derived || !devices[0].IsCurrent || + devices[0].Kind != sqlite.DeviceKindLocalInstallation || devices[0].CreatedAtMS != 1_910_000_000_000 { + t.Fatalf("google devices = %+v, want one current local device %q created at the first read", devices, derived) + } + if got := mustListDevices(t, v2, signalAccountID); !reflect.DeepEqual(got, signalDevicesBefore) { + t.Fatalf("signal devices = %+v, want unchanged %+v", got, signalDevicesBefore) + } + for deviceID, conversationID := range map[string]string{derived: google.ConversationID, otherDeviceID: signal.ConversationID} { + cursor, err := v2.GetReadCursor(deviceID, conversationID) + if err != nil || cursor.LastReadAtMS != 1_910_000_000_500 || cursor.LastReadMessageID != nil { + t.Fatalf("cursor on %q = %+v, %v; want read at 1910000000500 with no message", deviceID, cursor, err) + } + } +} + +// TestMarkReadV2CursorProperties runs random mark-read sequences over random +// account, device, and cursor shapes. Invariants, checked after every call: +// - the touched cursor's read time is the max of its previous read time and +// every mark-read time so far (monotone: an older call is a no-op), and a +// call that does not lose that race leaves exactly +// {LastReadMessageID: nil, LastReadAtMS: at, UpdatedAtMS: at}; +// - the v2 id and the legacy alias of a thread write the same cursor; +// - no other cursor, conversation, or account changes, and the only device +// ever added is LocalInstallationDeviceID for an account that had no local +// installation device; +// - the final state does not depend on the order of the calls. +func TestMarkReadV2CursorProperties(t *testing.T) { + type thread struct { + accountID string + remoteID string + } + threads := []thread{ + {googleAccountID, "google-thread-a"}, + {googleAccountID, "google-thread-b"}, + {whatsappAccountID, "15550001111@s.whatsapp.net"}, + {signalAccountID, "signal:+15550002222"}, + {signalAccountID, "signal-group:QUJD="}, + } + type call struct { + thread int + useAlias bool + atMS int64 + } + const ( + deviceNone = iota + deviceDerivedCurrent + deviceOtherNonCurrent + deviceShapeCount + ) + // build seeds the same store for a seed every time it is called, so a + // replay starts from an identical state. + build := func(seed int64) (*sqlite.Store, map[string]int, []sqlite.Conversation) { + random := rand.New(rand.NewSource(seed)) + v2 := openV2TestStore(t) + shapes := map[string]int{} + for _, accountID := range []string{googleAccountID, whatsappAccountID, signalAccountID} { + seedMarkReadAccount(t, v2, accountID) + shapes[accountID] = random.Intn(deviceShapeCount) + switch shapes[accountID] { + case deviceDerivedCurrent: + seedMarkReadDevice(t, v2, accountID, v2keys.LocalInstallationDeviceID(accountID), true) + case deviceOtherNonCurrent: + seedMarkReadDevice(t, v2, accountID, "other-local:"+accountID, false) + } + } + conversations := make([]sqlite.Conversation, len(threads)) + for index, thread := range threads { + conversations[index] = seedMarkReadConversation(t, v2, thread.accountID, thread.remoteID) + if shapes[thread.accountID] == deviceNone || random.Intn(2) == 0 { + continue + } + // A positioned cursor like the migration's or an ingested + // receipt's, at a time that may beat later mark-reads. + messageID := fmt.Sprintf("message-%d-%d", seed, index) + projectV2TestMessage(t, v2, sqlite.Message{ + MessageID: messageID, ConversationID: conversations[index].ConversationID, + AccountID: thread.accountID, RemoteMessageID: "remote-" + messageID, + Direction: sqlite.MessageDirectionOutgoing, Body: "seeded", State: sqlite.MessageStateActive, + OccurredAtMS: 1_900_000_000_000, + }) + device, err := v2.GetLocalInstallationDevice(context.Background(), thread.accountID) + if err != nil { + t.Fatalf("GetLocalInstallationDevice(): %v", err) + } + at := 1_910_000_000_000 + int64(random.Intn(2_000)) + if err := v2.UpsertReadCursor(sqlite.ReadCursor{ + AccountID: thread.accountID, DeviceID: device.DeviceID, + ConversationID: conversations[index].ConversationID, LastReadMessageID: &messageID, + LastReadAtMS: at, UpdatedAtMS: at, + }); err != nil { + t.Fatalf("UpsertReadCursor(seed): %v", err) + } + } + return v2, shapes, conversations + } + apply := func(v2 *sqlite.Store, conversations []sqlite.Conversation, c call) error { + key := conversations[c.thread].ConversationID + if c.useAlias { + key = threads[c.thread].remoteID + } + return MarkReadV2(context.Background(), v2, key, c.atMS) + } + + property := func(seed int64) bool { + random := rand.New(rand.NewSource(seed)) + v2, shapes, conversations := build(seed) + initial := snapshotV2(t, v2) + calls := make([]call, 1+random.Intn(12)) + for index := range calls { + // A narrow time window makes older, equal, and newer calls all + // common, against each other and against the seeded cursors. + calls[index] = call{ + thread: random.Intn(len(threads)), + useAlias: random.Intn(2) == 0, + atMS: 1_910_000_000_000 + int64(random.Intn(2_000)), + } + } + + model := initial + model.devices = map[string][]sqlite.Device{} + for accountID, devices := range initial.devices { + model.devices[accountID] = slices.Clone(devices) + } + model.cursors = cloneCursors(initial.cursors) + for step, c := range calls { + if err := apply(v2, conversations, c); err != nil { + t.Errorf("seed %d step %d: MarkReadV2(%+v): %v", seed, step, c, err) + return false + } + thread := threads[c.thread] + device, err := v2.GetLocalInstallationDevice(context.Background(), thread.accountID) + if err != nil { + t.Errorf("seed %d: GetLocalInstallationDevice(%q): %v", seed, thread.accountID, err) + return false + } + if shapes[thread.accountID] == deviceNone { + if device.DeviceID != v2keys.LocalInstallationDeviceID(thread.accountID) { + t.Errorf("seed %d: created device %q, want derived id", seed, device.DeviceID) + return false + } + if !containsDevice(model.devices[thread.accountID], device.DeviceID) { + model.devices[thread.accountID] = append(model.devices[thread.accountID], device) + } + } + key := cursorKey{device.DeviceID, conversations[c.thread].ConversationID} + if previous, ok := model.cursors[key]; !ok || c.atMS >= previous.LastReadAtMS { + model.cursors[key] = sqlite.ReadCursor{ + AccountID: thread.accountID, DeviceID: device.DeviceID, + ConversationID: key.conversationID, LastReadAtMS: c.atMS, UpdatedAtMS: c.atMS, + } + } + got := snapshotV2(t, v2) + if !reflect.DeepEqual(got, model) { + t.Errorf("seed %d step %d (%+v):\ngot %+v\nwant %+v", seed, step, c, got, model) + return false + } + } + + // Order independence: the same calls in reverse from the same start + // end at the same cursors. (A created device's timestamps record + // the first call, so only its ID is order independent.) + replay, _, replayConversations := build(seed) + for index := len(calls) - 1; index >= 0; index-- { + if err := apply(replay, replayConversations, calls[index]); err != nil { + t.Errorf("seed %d replay: %v", seed, err) + return false + } + } + reversed, forward := snapshotV2(t, replay), snapshotV2(t, v2) + if !reflect.DeepEqual(reversed.cursors, forward.cursors) { + t.Errorf("seed %d: reversed calls ended at\n%+v\nforward calls at\n%+v", seed, reversed.cursors, forward.cursors) + return false + } + for accountID, devices := range forward.devices { + if !reflect.DeepEqual(deviceIDs(reversed.devices[accountID]), deviceIDs(devices)) { + t.Errorf("seed %d: reversed devices %+v, forward %+v", seed, reversed.devices[accountID], devices) + return false + } + } + return true + } + if err := quick.Check(property, &quick.Config{MaxCount: 60, Rand: rand.New(rand.NewSource(20261010))}); err != nil { + t.Fatal(err) + } +} + +type cursorKey struct { + deviceID string + conversationID string +} + +// v2Snapshot is every row a mark-read could write: accounts, each account's +// devices, conversations, and the cursors of every device on every +// conversation. A cursor's foreign keys need an existing device and +// conversation, so a write the snapshot misses would show as a new one of +// those. +type v2Snapshot struct { + accounts []sqlite.Account + devices map[string][]sqlite.Device + conversations []sqlite.Conversation + cursors map[cursorKey]sqlite.ReadCursor +} + +func snapshotV2(t *testing.T, v2 *sqlite.Store) v2Snapshot { + t.Helper() + accounts, err := v2.ListAccounts() + if err != nil { + t.Fatalf("ListAccounts(): %v", err) + } + conversations, err := v2.ListConversationsByRecencyAllAccounts(10_000) + if err != nil { + t.Fatalf("ListConversationsByRecencyAllAccounts(): %v", err) + } + snapshot := v2Snapshot{ + accounts: accounts, + devices: map[string][]sqlite.Device{}, + conversations: conversations, + cursors: map[cursorKey]sqlite.ReadCursor{}, + } + for _, account := range accounts { + devices := mustListDevices(t, v2, account.AccountID) + snapshot.devices[account.AccountID] = devices + for _, device := range devices { + for _, conversation := range conversations { + cursor, err := v2.GetReadCursor(device.DeviceID, conversation.ConversationID) + if errors.Is(err, sqlite.ErrNotFound) { + continue + } + if err != nil { + t.Fatalf("GetReadCursor(): %v", err) + } + snapshot.cursors[cursorKey{device.DeviceID, conversation.ConversationID}] = cursor + } + } + } + return snapshot +} + +func deviceIDs(devices []sqlite.Device) []string { + ids := make([]string, 0, len(devices)) + for _, device := range devices { + ids = append(ids, device.DeviceID) + } + return ids +} + +func cloneCursors(cursors map[cursorKey]sqlite.ReadCursor) map[cursorKey]sqlite.ReadCursor { + clone := make(map[cursorKey]sqlite.ReadCursor, len(cursors)) + for key, cursor := range cursors { + clone[key] = cursor + } + return clone +} + +func mustMigratedConversation(t *testing.T, v2 *sqlite.Store, conversation migratedFixtureConversation) sqlite.Conversation { + t.Helper() + remoteID := v2keys.NormalizeRemoteConversationID(conversation.platform, conversation.legacyID) + owner, err := v2.GetConversationByRemote(conversation.accountID, remoteID) + if err != nil { + t.Fatalf("GetConversationByRemote(%q): %v", remoteID, err) + } + if owner.ConversationID == conversation.legacyID { + t.Fatalf("migrated conversation kept legacy ID %q; fixture no longer exercises a derived key", owner.ConversationID) + } + return owner +} + +func seedMarkReadAccount(t *testing.T, v2 *sqlite.Store, accountID string) { + t.Helper() + bridgeKey, displayName := accountBootstrap(accountID) + if err := v2.UpsertAccount(sqlite.Account{ + AccountID: accountID, BridgeKey: bridgeKey, DisplayName: displayName, + Mode: sqlite.AccountModeLive, Enabled: true, ConfigJSON: "{}", + CreatedAtMS: 1_800_000_000_000, UpdatedAtMS: 1_800_000_000_000, + }); err != nil { + t.Fatalf("UpsertAccount(%q): %v", accountID, err) + } +} + +func seedMarkReadDevice(t *testing.T, v2 *sqlite.Store, accountID, deviceID string, current bool) { + t.Helper() + if err := v2.UpsertDevice(sqlite.Device{ + DeviceID: deviceID, AccountID: accountID, Kind: sqlite.DeviceKindLocalInstallation, + DisplayName: "OpenMessage", State: sqlite.DeviceStateActive, IsCurrent: current, + CreatedAtMS: 1_800_000_000_000, UpdatedAtMS: 1_800_000_000_000, + }); err != nil { + t.Fatalf("UpsertDevice(%q): %v", deviceID, err) + } +} + +// seedMarkReadConversation stores a conversation the way live v2 ingest keys it: +// a derived ID, with the legacy-form remote ID as its natural key. +func seedMarkReadConversation(t *testing.T, v2 *sqlite.Store, accountID, remoteID string) sqlite.Conversation { + t.Helper() + conversation := sqlite.Conversation{ + ConversationID: v2keys.DeriveID("conversation", accountID, remoteID), + AccountID: accountID, + RemoteConversationID: remoteID, + Kind: sqlite.ConversationKindDirect, + NotificationMode: sqlite.NotificationModeAll, + MetadataJSON: "{}", + LastMessageAtMS: 1_900_000_000_000, + CreatedAtMS: 1_800_000_000_000, + UpdatedAtMS: 1_800_000_000_000, + } + if err := v2.UpsertConversation(conversation); err != nil { + t.Fatalf("UpsertConversation(%q): %v", remoteID, err) + } + stored, err := v2.GetConversation(conversation.ConversationID) + if err != nil { + t.Fatalf("GetConversation(%q): %v", conversation.ConversationID, err) + } + return stored +} diff --git a/internal/web/api.go b/internal/web/api.go index e8b27f89..544fd66e 100644 --- a/internal/web/api.go +++ b/internal/web/api.go @@ -35,6 +35,7 @@ import ( "github.com/maxghenis/openmessage/internal/readsource" "github.com/maxghenis/openmessage/internal/storage/sqlite" "github.com/maxghenis/openmessage/internal/story" + "github.com/maxghenis/openmessage/internal/v2wire" "github.com/maxghenis/openmessage/internal/whatsapplive" ) @@ -2348,14 +2349,24 @@ func APIHandlerWithOptions(store *db.Store, cli *client.Client, logger zerolog.L httpError(w, "mark read: "+err.Error(), 500) return } - // The legacy mirror is the legacy-primary dual-write seam, like the - // /api/v1 send split on V2Primary. A v2-primary UI sends v2 - // conversation ids, which the mirror cannot resolve against the - // legacy store, and a legacy id for a thread the v2 store lacks - // would add a legacy-keyed conversation row to the primary store. - if opts.V2 != nil && !opts.V2Primary { + // The v2 read cursor is best effort: a failure is logged and the + // response stays 200. It splits on V2Primary like the /api/v1 sends. + // The legacy mirror cannot resolve the v2 ids a v2-primary UI sends, + // and for a legacy id the v2 store lacks it would add a legacy-keyed + // conversation row to the primary store, so a v2-primary daemon + // writes the cursor natively. The write outlives a client that hangs + // up, like the legacy write above. + if opts.V2 != nil { nowMS := time.Now().UnixMilli() - if err := mirrorV2ReadCursor(r.Context(), store, opts.V2.V2Store, req.ConversationID, nowMS); err != nil { + ctx := context.WithoutCancel(r.Context()) + if opts.V2Primary { + if err := v2wire.MarkReadV2(ctx, opts.V2.V2Store, req.ConversationID, nowMS); err != nil { + logger.Warn(). + Err(err). + Str("conv_id", req.ConversationID). + Msg("Failed to write v2 read cursor") + } + } else if err := mirrorV2ReadCursor(ctx, store, opts.V2.V2Store, req.ConversationID, nowMS); err != nil { logger.Warn(). Err(err). Str("conv_id", req.ConversationID). diff --git a/internal/web/apiv1_test.go b/internal/web/apiv1_test.go index 537293d3..956d3df2 100644 --- a/internal/web/apiv1_test.go +++ b/internal/web/apiv1_test.go @@ -442,7 +442,9 @@ func TestMarkReadMirrorsOntoMigratedLocalDevice(t *testing.T) { // TestMarkReadSkipsLegacyMirrorOnV2Primary pins that a v2-primary daemon does // not run the legacy mirror: it would resolve the id against the legacy store // and, for a thread the v2 store lacks, add a legacy-keyed conversation (and -// the account and device rows around it) to the primary store. +// the account and device rows around it) to the primary store. The native +// write resolves only threads the v2 store holds, so it writes nothing here +// (markread_v2primary_test.go covers the threads it does hold). func TestMarkReadSkipsLegacyMirrorOnV2Primary(t *testing.T) { v2Store, err := sqlite.Open(filepath.Join(t.TempDir(), "v2.sqlite3")) if err != nil { diff --git a/internal/web/markread_v2primary_test.go b/internal/web/markread_v2primary_test.go new file mode 100644 index 00000000..62b09f4b --- /dev/null +++ b/internal/web/markread_v2primary_test.go @@ -0,0 +1,423 @@ +package web + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "io" + "math/rand" + "net/http" + "net/http/httptest" + "path/filepath" + "reflect" + "strings" + "testing" + "testing/quick" + "time" + + "github.com/rs/zerolog" + + "github.com/maxghenis/openmessage/internal/db" + "github.com/maxghenis/openmessage/internal/migration" + "github.com/maxghenis/openmessage/internal/storage/sqlite" + "github.com/maxghenis/openmessage/internal/v2keys" +) + +// markReadFixtureThread is a legacy conversation the fixture store held when +// it was migrated, so the v2 store keys it by derived hash and keeps the +// legacy id as its remote conversation id. +type markReadFixtureThread struct { + legacyID string + platform string + accountID string +} + +var markReadFixtureThreads = []markReadFixtureThread{ + {legacyID: "google-thread-1", platform: "sms", accountID: "google-primary"}, + {legacyID: "whatsapp:15550002222@s.whatsapp.net", platform: "whatsapp", accountID: "whatsapp-primary"}, + {legacyID: "signal:+15550003333", platform: "signal", accountID: "signal-primary"}, +} + +// TestMarkReadV2PrimaryWritesNativeCursorOnMigratedStore is the live +// v2-primary shape: the migration keyed each conversation and each account's +// local device by hash. The UI's v2 id and a stored legacy id both mark the +// same thread, on the migrated device, with no other v2 row touched. +func TestMarkReadV2PrimaryWritesNativeCursorOnMigratedStore(t *testing.T) { + v2 := openMigratedMarkReadStore(t) + handler := newMarkReadHandler(t, v2, zerolog.Nop()) + for _, thread := range markReadFixtureThreads { + owner := mustMigratedThread(t, v2, thread) + deviceID := mustMigratedLocalDevice(t, v2, thread.accountID) + for name, key := range map[string]string{"v2 id": owner.ConversationID, "legacy alias": thread.legacyID} { + t.Run(thread.legacyID+"/"+name, func(t *testing.T) { + before := snapshotMarkReadRows(t, v2) + previous := before.cursors[markReadCursorKey{deviceID, owner.ConversationID}] + + startMS := time.Now().UnixMilli() + postMarkRead(t, handler, key) + endMS := time.Now().UnixMilli() + + cursor, err := v2.GetReadCursor(deviceID, owner.ConversationID) + if err != nil { + t.Fatalf("GetReadCursor(migrated device %q): %v", deviceID, err) + } + if cursor.AccountID != thread.accountID || cursor.ConversationID != owner.ConversationID || + cursor.LastReadMessageID != nil || cursor.SourceUpdatedAtMS != nil || + cursor.LastReadAtMS < startMS || cursor.LastReadAtMS > endMS || + cursor.UpdatedAtMS != cursor.LastReadAtMS || cursor.LastReadAtMS < previous.LastReadAtMS { + t.Fatalf("cursor = %+v, want %s read in [%d, %d] with no message, after %+v", + cursor, owner.ConversationID, startMS, endMS, previous) + } + + after := snapshotMarkReadRows(t, v2) + delete(before.cursors, markReadCursorKey{deviceID, owner.ConversationID}) + delete(after.cursors, markReadCursorKey{deviceID, owner.ConversationID}) + if !reflect.DeepEqual(after, before) { + t.Fatalf("v2 rows besides the cursor changed:\nbefore %+v\nafter %+v", before, after) + } + if _, err := v2.GetDevice("local-primary:" + thread.accountID); !errors.Is(err, sqlite.ErrNotFound) { + t.Fatalf("GetDevice(local-primary) error = %v, want ErrNotFound", err) + } + if _, err := v2.GetConversation(thread.legacyID); !errors.Is(err, sqlite.ErrNotFound) { + t.Fatalf("GetConversation(legacy id) error = %v, want ErrNotFound: mark-read keyed a row by the legacy id", err) + } + }) + } + } +} + +// TestMarkReadV2PrimaryUnknownConversationLogsAndWritesNothing covers a +// legacy id for a thread the v2 store lacks: the response stays 200, the +// failure is logged, and nothing is written (the legacy mirror would have +// created the thread under its legacy id). +func TestMarkReadV2PrimaryUnknownConversationLogsAndWritesNothing(t *testing.T) { + v2 := openMigratedMarkReadStore(t) + var logs bytes.Buffer + handler := newMarkReadHandler(t, v2, zerolog.New(&logs)) + before := snapshotMarkReadRows(t, v2) + + postMarkRead(t, handler, "google-after-cutover") + + if after := snapshotMarkReadRows(t, v2); !reflect.DeepEqual(after, before) { + t.Fatalf("v2 rows changed:\nbefore %+v\nafter %+v", before, after) + } + assertMarkReadWarning(t, logs.String(), "google-after-cutover") +} + +func TestMarkReadV2PrimaryWithoutV2StoreStays200(t *testing.T) { + legacy, err := db.New(":memory:") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = legacy.Close() }) + if err := legacy.UpsertConversation(&db.Conversation{ + ConversationID: "google-thread-1", SourcePlatform: "sms", UnreadCount: 2, + }); err != nil { + t.Fatal(err) + } + var logs bytes.Buffer + handler := APIHandlerWithOptions(legacy, nil, zerolog.New(&logs), nil, APIOptions{ + V2Primary: true, + V2: &V2Options{}, + }) + + postMarkRead(t, handler, "google-thread-1") + + assertMarkReadWarning(t, logs.String(), "google-thread-1") + conversation, err := legacy.GetConversation("google-thread-1") + if err != nil { + t.Fatal(err) + } + if conversation.UnreadCount != 0 { + t.Fatalf("legacy unread count = %d, want 0", conversation.UnreadCount) + } +} + +// TestMarkReadV2PrimaryCursorMonotoneProperty drives the handler with random +// sequences of v2 ids and legacy aliases over a migrated store, after giving +// random threads a cursor from another writer (an ingested receipt or the +// migration) that is positioned on a message and dated in the past or the +// future. After every request: +// - a cursor dated after the request is unchanged (the write is monotone in +// read time, so a mark-read never moves a cursor back); +// - otherwise the cursor is {LastReadMessageID: nil, LastReadAtMS: +// UpdatedAtMS: the request's time}, never earlier than before; +// - no other cursor, device, account, or conversation changes. +func TestMarkReadV2PrimaryCursorMonotoneProperty(t *testing.T) { + property := func(seed int64) bool { + random := rand.New(rand.NewSource(seed)) + v2 := openMigratedMarkReadStore(t) + handler := newMarkReadHandler(t, v2, zerolog.Nop()) + messages, err := sqlite.NewMessageRepository(v2, time.Now) + if err != nil { + t.Fatal(err) + } + owners := make([]sqlite.Conversation, len(markReadFixtureThreads)) + devices := make([]string, len(markReadFixtureThreads)) + nowMS := time.Now().UnixMilli() + for index, thread := range markReadFixtureThreads { + owners[index] = mustMigratedThread(t, v2, thread) + devices[index] = mustMigratedLocalDevice(t, v2, thread.accountID) + if random.Intn(3) == 0 { + continue + } + latest, err := messages.ListMessagesByConversation(context.Background(), owners[index].ConversationID, 0, "", 1) + if err != nil || len(latest) != 1 { + t.Fatalf("latest message for %q = %+v, %v", owners[index].ConversationID, latest, err) + } + // Up to a day either side of now: past cursors lose to the next + // mark-read, future ones win against every request in the run. + at := nowMS + int64(random.Intn(2*86_400_000)) - 86_400_000 + if err := v2.UpsertReadCursor(sqlite.ReadCursor{ + AccountID: thread.accountID, DeviceID: devices[index], ConversationID: owners[index].ConversationID, + LastReadMessageID: &latest[0].MessageID, LastReadAtMS: at, UpdatedAtMS: at, + }); err != nil { + t.Fatalf("seed cursor: %v", err) + } + } + + for step := range 1 + random.Intn(8) { + index := random.Intn(len(markReadFixtureThreads)) + key := owners[index].ConversationID + if random.Intn(2) == 0 { + key = markReadFixtureThreads[index].legacyID + } + before := snapshotMarkReadRows(t, v2) + cursorKey := markReadCursorKey{devices[index], owners[index].ConversationID} + previous, hadCursor := before.cursors[cursorKey] + + startMS := time.Now().UnixMilli() + postMarkRead(t, handler, key) + endMS := time.Now().UnixMilli() + + after := snapshotMarkReadRows(t, v2) + cursor := after.cursors[cursorKey] + // The request reads the clock somewhere in [startMS, endMS]. A + // cursor dated after endMS must survive it, one dated at or before + // startMS must give way, and one dated inside the window may do + // either. + kept := hadCursor && reflect.DeepEqual(cursor, previous) && previous.LastReadAtMS > startMS + replaced := cursor.LastReadMessageID == nil && cursor.SourceUpdatedAtMS == nil && + cursor.LastReadAtMS >= startMS && cursor.LastReadAtMS <= endMS && + cursor.UpdatedAtMS == cursor.LastReadAtMS && + (!hadCursor || (cursor.LastReadAtMS >= previous.LastReadAtMS && previous.LastReadAtMS <= endMS)) + if !kept && !replaced { + t.Errorf("seed %d step %d: cursor %+v after %+v; want it kept if dated after the request, else read in [%d, %d] with no message", + seed, step, cursor, previous, startMS, endMS) + return false + } + delete(before.cursors, cursorKey) + delete(after.cursors, cursorKey) + if !reflect.DeepEqual(after, before) { + t.Errorf("seed %d step %d: other v2 rows changed:\nbefore %+v\nafter %+v", seed, step, before, after) + return false + } + } + return true + } + if err := quick.Check(property, &quick.Config{MaxCount: 15, Rand: rand.New(rand.NewSource(20261010))}); err != nil { + t.Fatal(err) + } +} + +// openMigratedMarkReadStore runs the real migration over a small legacy store, +// as internal/v2wire's openMigratedTestStores does, so the v2 store has the +// live install's shape: per-account accounts and a current local +// installation device per account under a derived ID, conversations keyed by +// derived hash, and a positioned read cursor per conversation. +func openMigratedMarkReadStore(t *testing.T) *sqlite.Store { + t.Helper() + root := t.TempDir() + legacyPath := filepath.Join(root, "messages.db") + legacy, err := db.New(legacyPath) + if err != nil { + t.Fatalf("db.New(): %v", err) + } + const baseMS int64 = 1_700_000_000_000 + for index, thread := range markReadFixtureThreads { + if err := legacy.UpsertConversation(&db.Conversation{ + ConversationID: thread.legacyID, + Name: "Migrated " + thread.platform, + Participants: `[]`, + LastMessageTS: baseMS + 10 + int64(index), + SourcePlatform: thread.platform, + UnreadCount: 1, + }); err != nil { + t.Fatalf("legacy UpsertConversation(%q): %v", thread.legacyID, err) + } + } + for _, message := range []*db.Message{ + {MessageID: "google-message-1", ConversationID: "google-thread-1", SenderNumber: "+15550001111", Body: "google 1", TimestampMS: baseMS, SourcePlatform: "sms"}, + {MessageID: "google-message-2", ConversationID: "google-thread-1", SenderNumber: "+15550001111", Body: "google 2", TimestampMS: baseMS + 10, SourcePlatform: "sms"}, + {MessageID: "whatsapp:wa-1", SourceID: "wa-1", ConversationID: "whatsapp:15550002222@s.whatsapp.net", SenderNumber: "15550002222@s.whatsapp.net", Body: "whatsapp 1", TimestampMS: baseMS + 1, SourcePlatform: "whatsapp"}, + {MessageID: "whatsapp:wa-2", SourceID: "wa-2", ConversationID: "whatsapp:15550002222@s.whatsapp.net", SenderNumber: "15550002222@s.whatsapp.net", Body: "whatsapp 2", TimestampMS: baseMS + 11, SourcePlatform: "whatsapp"}, + {MessageID: "signal:1700000000002", SourceID: "1700000000002", ConversationID: "signal:+15550003333", SenderNumber: "+15550003333", Body: "signal 1", TimestampMS: baseMS + 2, SourcePlatform: "signal"}, + {MessageID: "signal:1700000000012", SourceID: "1700000000012", ConversationID: "signal:+15550003333", SenderNumber: "+15550003333", Body: "signal 2", TimestampMS: baseMS + 12, SourcePlatform: "signal"}, + } { + if err := legacy.UpsertMessage(message); err != nil { + t.Fatalf("legacy UpsertMessage(%q): %v", message.MessageID, err) + } + } + if err := legacy.Close(); err != nil { + t.Fatalf("close legacy store: %v", err) + } + + staged := filepath.Join(root, "staged.sqlite3") + report, err := migration.Transform(context.Background(), migration.Options{ + SourcePath: legacyPath, + TempStorePath: staged, + TempBlobPath: filepath.Join(root, "blobs"), + TargetPath: filepath.Join(root, "target"), + TargetStorePath: filepath.Join(root, "target", "store.sqlite3"), + Check: true, + }) + if err != nil { + t.Fatalf("migration.Transform(): %v (report=%+v)", err, report) + } + v2, err := sqlite.Open(staged) + if err != nil { + t.Fatalf("sqlite.Open(migrated): %v", err) + } + t.Cleanup(func() { _ = v2.Close() }) + + // Pin the shape the tests depend on, so a migration change cannot make + // them pass vacuously: a derived device, and a migrated cursor that names + // a message (unread_count 1 leaves the first of two messages read). + for _, thread := range markReadFixtureThreads { + owner := mustMigratedThread(t, v2, thread) + deviceID := mustMigratedLocalDevice(t, v2, thread.accountID) + cursor, err := v2.GetReadCursor(deviceID, owner.ConversationID) + if err != nil || cursor.LastReadMessageID == nil { + t.Fatalf("migrated cursor for %q = %+v, %v; want one that names a message", thread.legacyID, cursor, err) + } + } + return v2 +} + +func mustMigratedThread(t *testing.T, v2 *sqlite.Store, thread markReadFixtureThread) sqlite.Conversation { + t.Helper() + remoteID := v2keys.NormalizeRemoteConversationID(thread.platform, thread.legacyID) + owner, err := v2.GetConversationByRemote(thread.accountID, remoteID) + if err != nil { + t.Fatalf("GetConversationByRemote(%q, %q): %v", thread.accountID, remoteID, err) + } + if owner.ConversationID == thread.legacyID { + t.Fatalf("migrated conversation kept legacy ID %q; fixture no longer exercises a derived key", owner.ConversationID) + } + return owner +} + +// mustMigratedLocalDevice returns the account's local installation device and +// checks it carries the migration's derived ID, not "local-primary:". +func mustMigratedLocalDevice(t *testing.T, v2 *sqlite.Store, accountID string) string { + t.Helper() + device, err := v2.GetLocalInstallationDevice(context.Background(), accountID) + if err != nil { + t.Fatalf("GetLocalInstallationDevice(%q): %v", accountID, err) + } + if want := v2keys.DeriveID("device", accountID, accountID+"\x1flocal"); device.DeviceID != want || !device.IsCurrent { + t.Fatalf("migrated local device for %q = %+v, want current device %q", accountID, device, want) + } + return device.DeviceID +} + +func newMarkReadHandler(t *testing.T, v2 *sqlite.Store, logger zerolog.Logger) http.Handler { + t.Helper() + legacy, err := db.New(":memory:") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = legacy.Close() }) + return APIHandlerWithOptions(legacy, nil, logger, nil, APIOptions{ + V2Primary: true, + V2: &V2Options{V2Store: v2}, + }) +} + +func postMarkRead(t *testing.T, handler http.Handler, conversationID string) { + t.Helper() + body, err := json.Marshal(map[string]string{"conversation_id": conversationID}) + if err != nil { + t.Fatal(err) + } + request := httptest.NewRequest(http.MethodPost, "http://127.0.0.1/api/mark-read", bytes.NewReader(body)) + request.Header.Set("Content-Type", "application/json") + recorder := httptest.NewRecorder() + handler.ServeHTTP(recorder, request) + response := recorder.Result() + defer response.Body.Close() + raw, _ := io.ReadAll(response.Body) + if response.StatusCode != http.StatusOK || !strings.Contains(string(raw), `"status":"ok"`) { + t.Fatalf("mark-read %q = %d %s, want 200 ok", conversationID, response.StatusCode, raw) + } +} + +func assertMarkReadWarning(t *testing.T, logs, conversationID string) { + t.Helper() + for _, line := range strings.Split(strings.TrimSpace(logs), "\n") { + var entry map[string]any + if json.Unmarshal([]byte(line), &entry) != nil { + continue + } + if entry["level"] == "warn" && entry["message"] == "Failed to write v2 read cursor" && + entry["conv_id"] == conversationID && entry["error"] != nil { + return + } + } + t.Fatalf("logs = %q, want a warning for %q", logs, conversationID) +} + +type markReadCursorKey struct { + deviceID string + conversationID string +} + +// markReadRows is every v2 row a mark-read could write: accounts, each +// account's devices, conversations, and every device's cursor on every +// conversation. A cursor's foreign keys need an existing device and +// conversation, so a write the snapshot misses would show as a new one. +type markReadRows struct { + accounts []sqlite.Account + devices map[string][]sqlite.Device + conversations []sqlite.Conversation + cursors map[markReadCursorKey]sqlite.ReadCursor +} + +func snapshotMarkReadRows(t *testing.T, v2 *sqlite.Store) markReadRows { + t.Helper() + accounts, err := v2.ListAccounts() + if err != nil { + t.Fatalf("ListAccounts(): %v", err) + } + conversations, err := v2.ListConversationsByRecencyAllAccounts(10_000) + if err != nil { + t.Fatalf("ListConversationsByRecencyAllAccounts(): %v", err) + } + rows := markReadRows{ + accounts: accounts, + devices: map[string][]sqlite.Device{}, + conversations: conversations, + cursors: map[markReadCursorKey]sqlite.ReadCursor{}, + } + for _, account := range accounts { + devices, err := v2.ListDevices(account.AccountID) + if err != nil { + t.Fatalf("ListDevices(%q): %v", account.AccountID, err) + } + rows.devices[account.AccountID] = devices + for _, device := range devices { + for _, conversation := range conversations { + cursor, err := v2.GetReadCursor(device.DeviceID, conversation.ConversationID) + if errors.Is(err, sqlite.ErrNotFound) { + continue + } + if err != nil { + t.Fatalf("GetReadCursor(): %v", err) + } + rows.cursors[markReadCursorKey{device.DeviceID, conversation.ConversationID}] = cursor + } + } + } + return rows +}