diff --git a/cmd/read.go b/cmd/read.go index 90565ada..880443bb 100644 --- a/cmd/read.go +++ b/cmd/read.go @@ -2,6 +2,7 @@ package cmd import ( "encoding/json" + "errors" "fmt" "io" "os" @@ -45,9 +46,11 @@ func openCommandReadSource( } if mode.Primary { storePath := filepath.Join(dataDir, "v2", "store.sqlite3") - store, err := sqlite.Open(storePath) + // A reader never migrates the store the app owns: a migration holds + // the write lock for its whole run and would stall the daemon's writes. + store, err := sqlite.OpenWithoutMigrating(storePath) if err != nil { - return nil, fmt.Errorf("open v2 read store %q: %w", storePath, err) + return nil, fmt.Errorf("open v2 read store %q: %w", storePath, clientStoreOpenError(err)) } if banner != nil { fmt.Fprintln(banner, "reading v2 store") @@ -173,3 +176,12 @@ func firstNonEmpty(vals ...string) string { func parseDayBound(s string, endOfDay bool) (int64, error) { return db.ParseDayBound(s, endOfDay) } + +// clientStoreOpenError explains the refusal a read client gets when this +// binary is newer than the store the app last upgraded. +func clientStoreOpenError(err error) error { + if errors.Is(err, sqlite.ErrMigrationPending) { + return fmt.Errorf("%w (quit and reopen the OpenMessage app so it upgrades the store, then retry)", err) + } + return err +} diff --git a/cmd/read_test.go b/cmd/read_test.go index 847888e5..c771c0a6 100644 --- a/cmd/read_test.go +++ b/cmd/read_test.go @@ -2,8 +2,13 @@ package cmd import ( "bytes" + "database/sql" + "errors" + "fmt" + "io" "os" "path/filepath" + "strings" "testing" "time" @@ -210,3 +215,78 @@ func TestOpenCommandReadSourceLegacyDoesNotRepairStore(t *testing.T) { t.Fatal("openCommandReadSource ran the startup repair sweeps: the legacy reaction placeholder was repaired away") } } + +// TestOpenCommandReadSourceNeverMigratesTheV2Store pins the other half of the +// read-only contract: `read` and `status` from a binary newer than the app's +// store refuse it instead of migrating it. A migration holds SQLite's write +// lock for its whole run, so applying one under the running app would stall +// (and, past busy_timeout, fail) the daemon's inbox appends. +func TestOpenCommandReadSourceNeverMigratesTheV2Store(t *testing.T) { + dataDir := t.TempDir() + t.Setenv("OPENMESSAGES_DATA_DIR", dataDir) + t.Setenv("OPENMESSAGES_DEMO", "0") + t.Setenv("OPENMESSAGES_APP_SANDBOX", "1") + t.Setenv("OPENMESSAGES_V2_PRIMARY", "1") + t.Setenv("OPENMESSAGES_V2_SEND", "") + t.Setenv("OPENMESSAGES_V2_INGEST", "") + + v2Dir := filepath.Join(dataDir, "v2") + if err := os.MkdirAll(v2Dir, 0o700); err != nil { + t.Fatalf("create v2 dir: %v", err) + } + storePath := filepath.Join(v2Dir, "store.sqlite3") + store, err := sqlite.Open(storePath) + if err != nil { + t.Fatalf("sqlite.Open(): %v", err) + } + if err := store.Close(); err != nil { + t.Fatal(err) + } + latest := rollBackLedgerOneVersion(t, storePath) + + session, err := openCommandReadSource(zerolog.Nop(), io.Discard) + if err == nil { + session.Close() + t.Fatal("openCommandReadSource() opened a store with a pending migration; want a refusal") + } + if !errors.Is(err, sqlite.ErrMigrationPending) || !strings.Contains(err.Error(), "reopen the OpenMessage app") { + t.Fatalf("openCommandReadSource() error = %v, want ErrMigrationPending telling the user to reopen the app", err) + } + if got := ledgerVersionCount(t, storePath); got != latest-1 { + t.Fatalf("ledger rows = %d after the refused read, want %d: the reader applied a migration", got, latest-1) + } +} + +// rollBackLedgerOneVersion removes the newest ledger row and lowers +// user_version to match: the shape of a store the running app has not +// upgraded to this build's schema yet. It returns the version it removed. +func rollBackLedgerOneVersion(t *testing.T, storePath string) int { + t.Helper() + latest := ledgerVersionCount(t, storePath) + database, err := sql.Open("sqlite", storePath) + if err != nil { + t.Fatal(err) + } + defer database.Close() + if _, err := database.Exec(`DELETE FROM schema_migrations WHERE version = ?`, latest); err != nil { + t.Fatal(err) + } + if _, err := database.Exec(fmt.Sprintf(`PRAGMA user_version = %d`, latest-1)); err != nil { + t.Fatal(err) + } + return latest +} + +func ledgerVersionCount(t *testing.T, storePath string) int { + t.Helper() + database, err := sql.Open("sqlite", storePath) + if err != nil { + t.Fatal(err) + } + defer database.Close() + var rows int + if err := database.QueryRow(`SELECT COUNT(*) FROM schema_migrations`).Scan(&rows); err != nil { + t.Fatal(err) + } + return rows +} diff --git a/cmd/repair.go b/cmd/repair.go index e0bef8b9..6e197137 100644 --- a/cmd/repair.go +++ b/cmd/repair.go @@ -100,9 +100,17 @@ func runRepairGoogleIDSpace(ctx context.Context, logger zerolog.Logger, args []s defer lock.Close() } - store, err := sqlite.Open(storePath) + // Only an exclusive owner migrates the store: --apply holds the instance + // lock and has checked the daemon is down. A dry run may run beside the + // app, and a migration's write lock would stall the daemon's writes. + var store *sqlite.Store + if apply { + store, err = sqlite.Open(storePath) + } else { + store, err = sqlite.OpenWithoutMigrating(storePath) + } if err != nil { - return fmt.Errorf("open v2 store: %w", err) + return fmt.Errorf("open v2 store: %w", clientStoreOpenError(err)) } defer store.Close() messages, err := sqlite.NewMessageRepository(store, time.Now) @@ -115,11 +123,12 @@ func runRepairGoogleIDSpace(ctx context.Context, logger zerolog.Logger, args []s if _, err := os.Stat(reference); err != nil { return fmt.Errorf("reference store %q: %w", reference, err) } - referenceStore, err = sqlite.Open(reference) + var closeReference func() + referenceStore, closeReference, err = openReferenceCopy(reference) if err != nil { return fmt.Errorf("open reference store: %w", err) } - defer referenceStore.Close() + defer closeReference() } report, err := ingest.PlanGoogleIDSpaceRepair(ctx, store, messages, ingest.IDSpaceRepairOptions{ @@ -228,6 +237,30 @@ func daemonAnswers(probeURL string) bool { return response.StatusCode == http.StatusOK } +// openReferenceCopy opens a copy of the reference snapshot, migrated to this +// build's schema, and returns a func that closes it and removes the copy. The +// snapshot is often older than the build, and migrating it in place would +// rewrite the user's backup. +func openReferenceCopy(reference string) (*sqlite.Store, func(), error) { + dir, err := os.MkdirTemp("", "openmessage-repair-reference-") + if err != nil { + return nil, nil, err + } + if err := copyStoreFiles(reference, dir); err != nil { + _ = os.RemoveAll(dir) + return nil, nil, fmt.Errorf("copy reference store: %w", err) + } + store, err := sqlite.Open(filepath.Join(dir, filepath.Base(reference))) + if err != nil { + _ = os.RemoveAll(dir) + return nil, nil, err + } + return store, func() { + _ = store.Close() + _ = os.RemoveAll(dir) + }, nil +} + func copyStoreFiles(storePath, backupDir string) error { if err := os.MkdirAll(backupDir, 0o700); err != nil { return err diff --git a/cmd/repair_test.go b/cmd/repair_test.go new file mode 100644 index 00000000..ff6207e3 --- /dev/null +++ b/cmd/repair_test.go @@ -0,0 +1,104 @@ +package cmd + +import ( + "bytes" + "context" + "errors" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/rs/zerolog" + + "github.com/maxghenis/openmessage/internal/storage/sqlite" +) + +// seedRepairStore provisions a current v2 store with the Google account the +// repair plans for, and returns its path. +func seedRepairStore(t *testing.T) string { + t.Helper() + dataDir := t.TempDir() + t.Setenv("OPENMESSAGES_DATA_DIR", dataDir) + t.Setenv("OPENMESSAGES_DEMO", "0") + t.Setenv("OPENMESSAGES_APP_SANDBOX", "1") + v2Dir := filepath.Join(dataDir, "v2") + if err := os.MkdirAll(v2Dir, 0o700); err != nil { + t.Fatalf("create v2 dir: %v", err) + } + storePath := filepath.Join(v2Dir, "store.sqlite3") + store, err := sqlite.Open(storePath) + if err != nil { + t.Fatalf("sqlite.Open(): %v", err) + } + nowMS := time.Now().UnixMilli() + if err := store.UpsertAccount(sqlite.Account{ + AccountID: "google-primary", + BridgeKey: "google_messages", + Mode: sqlite.AccountModeLive, + Enabled: true, + ConfigJSON: "{}", + CreatedAtMS: nowMS, + UpdatedAtMS: nowMS, + }); err != nil { + store.Close() + t.Fatalf("UpsertAccount(): %v", err) + } + if err := store.Close(); err != nil { + t.Fatal(err) + } + return storePath +} + +// A dry run needs no exclusive ownership (no daemon probe, no instance lock), +// so it must never migrate: run beside the app, a migration's write lock +// would stall the daemon's writes. +func TestRepairDryRunNeverMigratesTheStore(t *testing.T) { + storePath := seedRepairStore(t) + latest := rollBackLedgerOneVersion(t, storePath) + + var stdout, stderr bytes.Buffer + err := runRepairGoogleIDSpace(context.Background(), zerolog.Nop(), []string{"--since", "1"}, &stdout, &stderr) + if !errors.Is(err, sqlite.ErrMigrationPending) || !strings.Contains(err.Error(), "reopen the OpenMessage app") { + t.Fatalf("dry run error = %v, want ErrMigrationPending telling the user to reopen the app", err) + } + if got := ledgerVersionCount(t, storePath); got != latest-1 { + t.Fatalf("ledger rows = %d after the dry run, want %d: the dry run applied a migration", got, latest-1) + } +} + +// The reference snapshot is the user's own backup. The repair reads a migrated +// copy of it and never writes the file itself, so a read-only snapshot works +// and stays byte-identical. +func TestRepairDryRunReadsACopyOfTheReferenceSnapshot(t *testing.T) { + storePath := seedRepairStore(t) + reference := filepath.Join(t.TempDir(), "reference.sqlite3") + payload, err := os.ReadFile(storePath) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(reference, payload, 0o400); err != nil { + t.Fatal(err) + } + + var stdout, stderr bytes.Buffer + if err := runRepairGoogleIDSpace( + context.Background(), zerolog.Nop(), + []string{"--since", "1", "--reference", reference, "--json"}, &stdout, &stderr, + ); err != nil { + t.Fatalf("dry run with a read-only reference: %v\n%s", err, stderr.String()) + } + after, err := os.ReadFile(reference) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(payload, after) { + t.Fatal("the repair wrote the reference snapshot") + } + for _, suffix := range []string{"-wal", "-shm"} { + if _, err := os.Stat(reference + suffix); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("the repair created %s next to the reference snapshot: %v", reference+suffix, err) + } + } +} diff --git a/cmd/serve.go b/cmd/serve.go index d2caffcc..0f4b94d5 100644 --- a/cmd/serve.go +++ b/cmd/serve.go @@ -881,9 +881,11 @@ func runServeMCPClient(logger zerolog.Logger, opts serveOptions) error { if _, err := os.Stat(v2StorePath); err != nil { return fmt.Errorf("v2-primary reads selected but no migrated store at %s: %w", v2StorePath, err) } - opened, err := sqlite.Open(v2StorePath) + // Never migrate here: the daemon owns the store, and a migration's + // write lock would stall its writes for the whole index build. + opened, err := sqlite.OpenWithoutMigrating(v2StorePath) if err != nil { - return fmt.Errorf("open v2 read store %q: %w", v2StorePath, err) + return fmt.Errorf("open v2 read store %q: %w", v2StorePath, clientStoreOpenError(err)) } v2Store = opened defer func() { diff --git a/cmd/serve_client_test.go b/cmd/serve_client_test.go index 53ec122a..14eecd3e 100644 --- a/cmd/serve_client_test.go +++ b/cmd/serve_client_test.go @@ -3,6 +3,7 @@ package cmd import ( "bytes" "encoding/json" + "errors" "net/http" "net/http/httptest" "net/url" @@ -204,6 +205,55 @@ func TestRunServeMCPClientAdoptsDaemonTruth(t *testing.T) { } } +// TestRunServeMCPClientNeverMigratesTheV2Store pins that the per-session MCP +// client refuses a store the running app has not upgraded instead of +// migrating it: the migration's write lock would stall the daemon's writes. +func TestRunServeMCPClientNeverMigratesTheV2Store(t *testing.T) { + dataDir := t.TempDir() + v2Dir := filepath.Join(dataDir, "v2") + if err := os.MkdirAll(v2Dir, 0o700); err != nil { + t.Fatalf("create v2 dir: %v", err) + } + storePath := filepath.Join(v2Dir, "store.sqlite3") + store, err := sqlite.Open(storePath) + if err != nil { + t.Fatalf("provision v2 store: %v", err) + } + if err := store.Close(); err != nil { + t.Fatalf("close provisioned v2 store: %v", err) + } + latest := rollBackLedgerOneVersion(t, storePath) + + daemon := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/status" { + http.NotFound(w, r) + return + } + json.NewEncoder(w).Encode(map[string]any{ + "connected": true, + "v2_primary": true, + "v2_send": true, + "auth": map[string]any{"data_dir": dataDir}, + }) + })) + defer daemon.Close() + daemonURL, err := url.Parse(daemon.URL) + if err != nil { + t.Fatalf("parse daemon URL: %v", err) + } + setClientModeTestEnv(t, dataDir) + t.Setenv("OPENMESSAGES_PORT", daemonURL.Port()) + + var logs syncBuffer + err = RunServe(zerolog.New(&logs), "--mcp-stdio") + if !errors.Is(err, sqlite.ErrMigrationPending) || !strings.Contains(err.Error(), "reopen the OpenMessage app") { + t.Fatalf("RunServe(--mcp-stdio) error = %v, want ErrMigrationPending telling the user to reopen the app\n%s", err, logs.String()) + } + if got := ledgerVersionCount(t, storePath); got != latest-1 { + t.Fatalf("ledger rows = %d after the refused client start, want %d: the client applied a migration", got, latest-1) + } +} + // TestRunServeMCPClientAdoptsDaemonDataDir verifies that with no // OPENMESSAGES_DATA_DIR pinned, the client adopts the data directory the // running daemon reports, so an env-less MCP spawn serves the app's store diff --git a/docs/agent-runbook.md b/docs/agent-runbook.md index e97698eb..cea9105b 100644 --- a/docs/agent-runbook.md +++ b/docs/agent-runbook.md @@ -438,6 +438,17 @@ running app: `TestNewClientPerformsNoStoreWrites` (internal/app), `TestRunServeMCPClientDoesNotRepairStore` and `TestOpenCommandReadSourceLegacyDoesNotRepairStore` (cmd). +- **The v2 store opens without migrating** (`sqlite.OpenWithoutMigrating`): + only the store's owner applies schema migrations, i.e. the daemon as the + first step of building its v2 stack, plus `repair` and cutover. A migration + holds SQLite's write lock for its whole run (index builds take seconds on a + large store) and the daemon's writers give up after the 5 s busy timeout, + so a client migrating under the running app would drop inbox appends. A + client binary newer than the store refuses it with "quit and reopen the + OpenMessage app"; one older than the store refuses it as "newer than + supported". Regression tests: `TestOpenWithoutMigrating*` + (internal/storage/sqlite), `TestOpenCommandReadSourceNeverMigratesTheV2Store` + and `TestRunServeMCPClientNeverMigratesTheV2Store` (cmd). - **Sends/reactions route through the daemon** (`/api/v1/outbox` on v2, `/api/send`+`/api/react` on legacy), like the CLI has done since PR #140, with the same do-not-resend idempotency contract. With the app closed, @@ -471,8 +482,8 @@ daemon-truth detection all line up (two-data-dirs trap above). On a migrated `messages.db` froze at cutover, and this keeps MCP reads on the v2 store even when the app is closed or predates the `auth.data_dir` status field (drop the line on a non-migrated install). Keep the PATH binary in lockstep with the -installed app — both open the same SQLite stores and a version-skewed binary -can migrate the schema under the older one. +installed app — both open the same SQLite stores, and a client binary at a +different schema version than the store refuses it until the two match. **Never** configure MCP to run `serve --web`, `serve --mcp-sse`, or `serve ... --transports` alongside the app: those are daemon shapes and will @@ -1497,6 +1508,12 @@ xattr -cr /Applications/OpenMessage.app open -a OpenMessage ``` +A build with a new schema migration applies it when the new app starts, +before its v2 stack (and so v2 ingest) exists. Update the PATH binary +(`/usr/local/bin/openmessage`) after the app: until the two match, `openmessage +read`/`status` and the MCP client refuse the store (a newer binary asks you to +reopen the app; an older one reports the store as newer than supported). + Confirm the deployed bundle kept the release id: ``` diff --git a/internal/db/messages.go b/internal/db/messages.go index 8c68d1b6..4bdf8748 100644 --- a/internal/db/messages.go +++ b/internal/db/messages.go @@ -780,6 +780,14 @@ type PlatformStat struct { LatestRecvMS int64 // most recent received message, 0 if none } +// PlatformLatest is a PlatformStat without the message count, for callers that +// only judge freshness and should not pay for counting every message. +type PlatformLatest struct { + Platform string + LatestMS int64 // most recent message (sent or received), 0 if none + LatestRecvMS int64 // most recent received message, 0 if none +} + // PlatformStats returns per-platform message counts and the latest sent/received // timestamps in a single GROUP BY pass, ordered most-recent-activity first. // 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..417632e7 100644 --- a/internal/migration/validate.go +++ b/internal/migration/validate.go @@ -23,7 +23,7 @@ var targetCountTables = []string{ "read_cursors", } -const migration0010Checksum = "dfab4551335d92045cb895e5b2c781f4f57216bbc428a705fc26bbdd474db0b1" +const migration0011Checksum = "a0b486411095e4b1960dba219a0eefbafc243363d450875ba93443fcc0e367f3" func checkpointAndSyncSQLite(ctx context.Context, path string) error { database, err := sql.Open("sqlite", path) @@ -151,8 +151,8 @@ 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.MigrationChecksums[9] == migration0010Checksum && + report.Target.SchemaVersion == 11 && len(report.Target.MigrationChecksums) == 11 && + 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/content_dedupe.go b/internal/storage/sqlite/content_dedupe.go index 68b60a17..7dccd858 100644 --- a/internal/storage/sqlite/content_dedupe.go +++ b/internal/storage/sqlite/content_dedupe.go @@ -7,6 +7,25 @@ import ( "fmt" ) +// messageContentDuplicateQuery runs for every Google message with a body. It +// must seek on the exact occurrence millisecond: through +// messages_conversation_time_idx, or messages_account_direction_time_idx, which +// the planner prefers for its third equality. Either reads only that +// millisecond's rows. +const messageContentDuplicateQuery = ` + SELECT ` + messageColumns + ` + FROM messages + WHERE account_id = ? + AND conversation_id = ? + AND occurred_at_ms = ? + AND direction = ? + AND body = ? + AND sender_identity_id IS ? + AND remote_message_id <> ? + ORDER BY created_at_ms, message_id + LIMIT 1 +` + // FindMessageContentDuplicate returns a message in the same conversation with // identical content (direction, sender, occurrence millisecond, and body) but // a different remote message ID. This is the signature of a re-delivery after @@ -24,19 +43,7 @@ func (r *MessageRepository) FindMessageContentDuplicate( occurredAtMS int64, body string, ) (Message, bool, error) { - message, err := scanMessage(r.store.db.QueryRowContext(ctx, ` - SELECT `+messageColumns+` - FROM messages - WHERE account_id = ? - AND conversation_id = ? - AND occurred_at_ms = ? - AND direction = ? - AND body = ? - AND sender_identity_id IS ? - AND remote_message_id <> ? - ORDER BY created_at_ms, message_id - LIMIT 1 - `, + message, err := scanMessage(r.store.db.QueryRowContext(ctx, messageContentDuplicateQuery, accountID, conversationID, occurredAtMS, 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_activity.go b/internal/storage/sqlite/message_activity.go new file mode 100644 index 00000000..0f070bec --- /dev/null +++ b/internal/storage/sqlite/message_activity.go @@ -0,0 +1,64 @@ +package sqlite + +import ( + "context" + "database/sql" + "fmt" +) + +// The per-account aggregates below replace v2read code that loaded every +// conversation of an account and then paged through every message (platform +// stats and message counts), read each conversation's newest message (latest +// timestamp), or counted the loaded rows (conversation counts). MAX on a +// literal direction is a single seek of messages_account_direction_time_idx; +// COUNT(*) is a covering range of an account-leading index, which never reads +// the message rows themselves. +const ( + latestMessageTimesQuery = ` + SELECT + (SELECT MAX(occurred_at_ms) FROM messages + WHERE account_id = ? AND direction = 'incoming'), + (SELECT MAX(occurred_at_ms) FROM messages + WHERE account_id = ? AND direction = 'outgoing') + ` + countMessagesQuery = `SELECT COUNT(*) FROM messages WHERE account_id = ?` + countConversationsQuery = `SELECT COUNT(*) FROM conversations WHERE account_id = ?` +) + +// LatestMessageTimes returns the occurred_at_ms of the account's newest +// message in either direction and of its newest incoming message. Each is 0 +// when the account has no such message. +func (s *Store) LatestMessageTimes(accountID string) (latestMS, latestIncomingMS int64, err error) { + var incoming, outgoing sql.NullInt64 + if err := s.db.QueryRowContext( + context.Background(), latestMessageTimesQuery, accountID, accountID, + ).Scan(&incoming, &outgoing); err != nil { + return 0, 0, fmt.Errorf("latest message times for account %q: %w", accountID, err) + } + latestMS = max(incoming.Int64, outgoing.Int64) + return latestMS, incoming.Int64, nil +} + +// CountMessages returns the number of stored messages of the account, in +// every direction and state. +func (s *Store) CountMessages(accountID string) (int64, error) { + var count int64 + if err := s.db.QueryRowContext( + context.Background(), countMessagesQuery, accountID, + ).Scan(&count); err != nil { + return 0, fmt.Errorf("count messages for account %q: %w", accountID, err) + } + return count, nil +} + +// CountConversations returns the number of conversations of the account, +// archived ones included. +func (s *Store) CountConversations(accountID string) (int64, error) { + var count int64 + if err := s.db.QueryRowContext( + context.Background(), countConversationsQuery, accountID, + ).Scan(&count); err != nil { + return 0, fmt.Errorf("count conversations for account %q: %w", accountID, err) + } + return count, nil +} 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/message_merge.go b/internal/storage/sqlite/message_merge.go index 25e81512..72672cb7 100644 --- a/internal/storage/sqlite/message_merge.go +++ b/internal/storage/sqlite/message_merge.go @@ -339,12 +339,9 @@ func messageLocation(ctx context.Context, tx *sql.Tx, messageID string) (string, // the message left its conversation. func readCursorDeviceOutside(ctx context.Context, tx *sql.Tx, messageID, conversationID string) (string, bool, error) { var deviceID string - err := tx.QueryRowContext(ctx, ` - SELECT device_id FROM read_cursors - WHERE last_read_message_id = ? AND conversation_id <> ? - ORDER BY device_id - LIMIT 1 - `, messageID, conversationID).Scan(&deviceID) + err := tx.QueryRowContext( + ctx, readCursorDeviceOutsideQuery, messageID, conversationID, messageID, + ).Scan(&deviceID) if errors.Is(err, sql.ErrNoRows) { return "", false, nil } @@ -353,3 +350,18 @@ func readCursorDeviceOutside(ctx context.Context, tx *sql.Tx, messageID, convers } return deviceID, true, nil } + +// readCursorDeviceOutsideQuery finds such a cursor without an index on +// last_read_message_id: the cursor's composite foreign key +// (conversation_id, last_read_message_id) -> messages(conversation_id, +// message_id) puts any cursor naming the message in the message's own +// conversation, so the lookup seeks read_cursors_conversation_idx there +// instead of scanning every cursor. +const readCursorDeviceOutsideQuery = ` + SELECT device_id FROM read_cursors + WHERE conversation_id = (SELECT conversation_id FROM messages WHERE message_id = ?) + AND conversation_id <> ? + AND last_read_message_id = ? + ORDER BY device_id + LIMIT 1 +` diff --git a/internal/storage/sqlite/messages.go b/internal/storage/sqlite/messages.go index 45131375..43336564 100644 --- a/internal/storage/sqlite/messages.go +++ b/internal/storage/sqlite/messages.go @@ -295,6 +295,32 @@ func (r *MessageRepository) GetMessageByRemote( return message, nil } +// Keyset pages compare (occurred_at_ms, message_id) as one row value, which +// SQLite turns into a range on messages_conversation_time_idx and so reads only +// the page. The equivalent OR form, +// occurred_at_ms < ? OR (occurred_at_ms = ? AND message_id < ?), bounds only +// conversation_id: each page then walks every row between the conversation's +// newest message and the cursor, and walking a whole thread is quadratic. +// Both columns are NOT NULL, so the two forms select the same rows. +const ( + messagesBeforeCursorQuery = ` + SELECT ` + messageColumns + ` + FROM messages + WHERE conversation_id = ? + AND (occurred_at_ms, message_id) < (?, ?) + ORDER BY occurred_at_ms DESC, message_id DESC + LIMIT ? + ` + messagesAfterCursorQuery = ` + SELECT ` + messageColumns + ` + FROM messages + WHERE conversation_id = ? + AND (occurred_at_ms, message_id) > (?, ?) + ORDER BY occurred_at_ms ASC, message_id ASC + LIMIT ? + ` +) + // ListMessagesByConversation returns a newest-first page. beforeMS == 0 // selects the latest page; otherwise beforeID is the deterministic tie cursor. func (r *MessageRepository) ListMessagesByConversation( @@ -325,15 +351,8 @@ func (r *MessageRepository) ListMessagesByConversation( ` args = []any{conversationID, beforeMS, limit} if beforeID != "" { - query = ` - SELECT ` + messageColumns + ` - FROM messages - WHERE conversation_id = ? - AND (occurred_at_ms < ? OR (occurred_at_ms = ? AND message_id < ?)) - ORDER BY occurred_at_ms DESC, message_id DESC - LIMIT ? - ` - args = []any{conversationID, beforeMS, beforeMS, beforeID, limit} + query = messagesBeforeCursorQuery + args = []any{conversationID, beforeMS, beforeID, limit} } } rows, err := r.store.db.QueryContext(ctx, query, args...) @@ -368,15 +387,8 @@ func (r *MessageRepository) ListMessagesByConversationAfter( ` args := []any{conversationID, afterMS, limit} if afterID != "" { - query = ` - SELECT ` + messageColumns + ` - FROM messages - WHERE conversation_id = ? - AND (occurred_at_ms > ? OR (occurred_at_ms = ? AND message_id > ?)) - ORDER BY occurred_at_ms ASC, message_id ASC - LIMIT ? - ` - args = []any{conversationID, afterMS, afterMS, afterID, limit} + query = messagesAfterCursorQuery + args = []any{conversationID, afterMS, afterID, limit} } rows, err := r.store.db.QueryContext(ctx, query, args...) if err != nil { @@ -420,14 +432,10 @@ func (r *MessageRepository) ListMessagesAroundMessage( return nil, fmt.Errorf("message %q in conversation %q: %w", messageID, conversationID, ErrNotFound) } - beforeRows, err := r.store.db.QueryContext(ctx, ` - SELECT `+messageColumns+` - FROM messages - WHERE conversation_id = ? - AND (occurred_at_ms < ? OR (occurred_at_ms = ? AND message_id < ?)) - ORDER BY occurred_at_ms DESC, message_id DESC - LIMIT ? - `, conversationID, anchor.OccurredAtMS, anchor.OccurredAtMS, anchor.MessageID, before) + beforeRows, err := r.store.db.QueryContext( + ctx, messagesBeforeCursorQuery, + conversationID, anchor.OccurredAtMS, anchor.MessageID, before, + ) if err != nil { return nil, fmt.Errorf("list messages before %q: %w", messageID, err) } @@ -436,14 +444,10 @@ func (r *MessageRepository) ListMessagesAroundMessage( return nil, fmt.Errorf("list messages before %q: %w", messageID, err) } - afterRows, err := r.store.db.QueryContext(ctx, ` - SELECT `+messageColumns+` - FROM messages - WHERE conversation_id = ? - AND (occurred_at_ms > ? OR (occurred_at_ms = ? AND message_id > ?)) - ORDER BY occurred_at_ms ASC, message_id ASC - LIMIT ? - `, conversationID, anchor.OccurredAtMS, anchor.OccurredAtMS, anchor.MessageID, after) + afterRows, err := r.store.db.QueryContext( + ctx, messagesAfterCursorQuery, + conversationID, anchor.OccurredAtMS, anchor.MessageID, after, + ) if err != nil { return nil, fmt.Errorf("list messages after %q: %w", messageID, err) } @@ -476,10 +480,36 @@ func (r *MessageRepository) SearchMessages( filter.SinceMS, filter.UntilMS = filter.UntilMS, filter.SinceMS } + statement, args := searchMessagesStatement(query, filter) + rows, err := r.store.db.QueryContext(ctx, statement, args...) + if err != nil { + return nil, fmt.Errorf("search messages: %w", err) + } + messages, err := collectRows(rows, scanMessage) + if err != nil { + return nil, fmt.Errorf("search messages: %w", err) + } + return messages, nil +} + +// searchMessagesStatement builds SearchMessages' SQL for an already +// normalized filter. +// +// The sender filter selects the sender's identity IDs in a subquery so the +// planner can seek messages_sender_time_idx per identity. Joining identities +// and filtering on i.canonical_value instead walked that whole index, every +// attributed message, because no identities index leads with canonical_value. +// The subquery needs no account_id match: the composite foreign key +// messages(account_id, sender_identity_id) -> identities(account_id, +// identity_id) already makes a sender identity belong to the message's account. +func searchMessagesStatement(query string, filter SearchQuery) (string, []any) { conditions := []string{"m.body LIKE '%' || ? || '%'"} args := []any{query} if filter.AccountID != "" { - conditions = append(conditions, "m.account_id = ?") + // direction's CHECK allows exactly these two values, so the IN changes + // no result; it lets messages_account_direction_time_idx bound a date + // window per direction instead of reading the account's every row. + conditions = append(conditions, "m.account_id = ? AND m.direction IN ('incoming', 'outgoing')") args = append(args, filter.AccountID) } if filter.ConversationID != "" { @@ -487,7 +517,9 @@ func (r *MessageRepository) SearchMessages( args = append(args, filter.ConversationID) } if filter.SenderCanonicalValue != "" { - conditions = append(conditions, "i.canonical_value = ?") + conditions = append(conditions, `m.sender_identity_id IN ( + SELECT identity_id FROM identities WHERE canonical_value = ? + )`) args = append(args, filter.SenderCanonicalValue) } if filter.SinceMS > 0 { @@ -500,24 +532,25 @@ func (r *MessageRepository) SearchMessages( } args = append(args, filter.Limit) + // A listing with no substring walks messages_time_idx from the newest end + // (of the window, if any) and stops after limit rows; filtered by account + // it reads that account's rows inside the window. A substring search that + // no conversation or sender bounds must not use an index: for a rare or + // absent term the planner would walk messages_time_idx and fetch each row + // out of rowid order, several times slower than one sequential scan. Pin + // that scan (with its top-N sort) until full-text search replaces LIKE. + from := "messages AS m" + if query != "" && filter.ConversationID == "" && filter.SenderCanonicalValue == "" { + from = "messages AS m NOT INDEXED" + } statement := ` SELECT ` + prefixedMessageColumns("m") + ` - FROM messages AS m - LEFT JOIN identities AS i - ON i.account_id = m.account_id AND i.identity_id = m.sender_identity_id + FROM ` + from + ` WHERE ` + strings.Join(conditions, " AND ") + ` ORDER BY m.occurred_at_ms DESC, m.message_id DESC LIMIT ? ` - rows, err := r.store.db.QueryContext(ctx, statement, args...) - if err != nil { - return nil, fmt.Errorf("search messages: %w", err) - } - messages, err := collectRows(rows, scanMessage) - if err != nil { - return nil, fmt.Errorf("search messages: %w", err) - } - return messages, nil + return statement, args } // ImportMessage atomically upserts a historical normalized message by remote 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 5865c082..045f756c 100644 --- a/internal/storage/sqlite/migrations.go +++ b/internal/storage/sqlite/migrations.go @@ -67,6 +67,9 @@ var migration0009SQL string //go:embed migrations/0010_reactions.sql var migration0010SQL string +//go:embed migrations/0011_read_path_indexes.sql +var migration0011SQL string + var embeddedMigrations = []migration{ newMigration(1, "storage_shell", migration0001SQL, newStorageShellArguments), newMigration(2, "identity_graph", migration0002SQL, nil), @@ -78,6 +81,7 @@ var embeddedMigrations = []migration{ newMigration(8, "message_attachments", migration0008SQL, nil), newMigration(9, "outbox_send_again", migration0009SQL, nil), newMigration(10, "reactions", migration0010SQL, nil), + newMigration(11, "read_path_indexes", migration0011SQL, nil), } func newMigration( diff --git a/internal/storage/sqlite/migrations/0011_read_path_indexes.sql b/internal/storage/sqlite/migrations/0011_read_path_indexes.sql new file mode 100644 index 00000000..c969c467 --- /dev/null +++ b/internal/storage/sqlite/migrations/0011_read_path_indexes.sql @@ -0,0 +1,25 @@ +-- Indexes that keep hot read paths off whole-table scans (plan audit, +-- 2026-10-09). Each serves a read that otherwise visits every row of a table +-- that grows with message history. + +-- /api/status refreshes every platform's newest message and newest incoming +-- message every 30 s while the app is open. MAX(occurred_at_ms) for one +-- (account_id, direction) is a single seek here. +CREATE INDEX messages_account_direction_time_idx + ON messages(account_id, direction, occurred_at_ms); + +-- Newest-first listings across conversations (MCP get_messages, a search with +-- an empty query, optionally inside a date window) read the first rows of this +-- index instead of sorting the whole table to keep a few. +CREATE INDEX messages_time_idx + ON messages(occurred_at_ms, message_id); + +-- Outbox rows by the local message they send. Every outgoing message a thread +-- renders asks for its newest outbox row (account_id and local_message_id +-- both bound), and repair asks whether a message has an unfinished send +-- (local_message_id alone); without this index each lookup reads every outbox +-- row of the account or of the store. local_message_id = ? implies the +-- partial predicate. +CREATE INDEX outbox_local_message_idx + ON outbox(local_message_id, account_id, created_at_ms, outbox_id) + WHERE local_message_id IS NOT NULL; diff --git a/internal/storage/sqlite/migrations_fk_test.go b/internal/storage/sqlite/migrations_fk_test.go index dc3c2d0d..a995cbec 100644 --- a/internal/storage/sqlite/migrations_fk_test.go +++ b/internal/storage/sqlite/migrations_fk_test.go @@ -70,6 +70,12 @@ CREATE INDEX messages_sender_time_idx CREATE UNIQUE INDEX messages_conversation_message_uq ON messages(conversation_id, message_id); + +CREATE INDEX messages_account_direction_time_idx + ON messages(account_id, direction, occurred_at_ms); + +CREATE INDEX messages_time_idx + ON messages(occurred_at_ms, message_id); ` // messageChildTables are every table with a foreign key to messages. diff --git a/internal/storage/sqlite/open_without_migrating.go b/internal/storage/sqlite/open_without_migrating.go new file mode 100644 index 00000000..c69a708d --- /dev/null +++ b/internal/storage/sqlite/open_without_migrating.go @@ -0,0 +1,105 @@ +package sqlite + +import ( + "context" + "database/sql" + "errors" + "fmt" + "net/url" + "os" + "strings" +) + +// ErrMigrationPending reports a store whose schema is older than this build's. +// Only a process that owns the store applies migrations, so a client refuses +// the store until the owner (the app's daemon, at its next start) upgrades it. +var ErrMigrationPending = errors.New("sqlite store has migrations this build would apply") + +// OpenWithoutMigrating opens an existing store that is already at this build's +// schema, for processes that read a store another process owns: openmessage +// read and status, and the transportless MCP client. Unlike Open it never +// creates the file and never applies a migration. A migration runs inside +// BEGIN IMMEDIATE and holds SQLite's write lock until it commits, and an index +// build on a large store takes longer than the owner's busy_timeout, so a +// client that migrated underneath the running daemon would make the daemon's +// writes (inbox appends among them) fail with SQLITE_BUSY. +// +// The ledger is checked in a read-only (deferred) transaction, which never +// takes the write reservation: opening neither waits for nor blocks the +// owner's writes. A store with pending migrations returns ErrMigrationPending; +// one migrated by a newer build, or whose ledger does not match this build's +// migrations, returns the same errors Open would. The connection is opened +// with mode=rw, so a file removed after the existence check is not recreated. +// It is otherwise an ordinary read-write connection (the fuller read-only +// attach is #166's OpenReadOnly): like any SQLite connection that closes last, +// it may checkpoint a WAL nobody else has open into the main file, which +// changes no row. +func OpenWithoutMigrating(path string) (*Store, error) { + return openWithoutMigrating(path, embeddedMigrations) +} + +func openWithoutMigrating(path string, migrations []migration) (*Store, error) { + if strings.TrimSpace(path) == "" { + return nil, fmt.Errorf("open sqlite store without migrating: path is empty") + } + if err := validateMigrationList(migrations); err != nil { + return nil, err + } + if _, err := os.Stat(path); err != nil { + return nil, fmt.Errorf("open sqlite store without migrating: %w", err) + } + + db, err := sql.Open("sqlite", existingStoreDSN(path)) + if err != nil { + return nil, fmt.Errorf("open sqlite store: %w", err) + } + ctx := context.Background() + if err := verifyConnectionPragmas(ctx, db); err != nil { + _ = db.Close() + return nil, err + } + if err := checkMigrated(ctx, db, migrations); err != nil { + _ = db.Close() + return nil, err + } + return &Store{db: db}, nil +} + +// checkMigrated validates the ledger against migrations without applying any. +func checkMigrated(ctx context.Context, db *sql.DB, migrations []migration) error { + tx, err := db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true}) + if err != nil { + return fmt.Errorf("begin read-only ledger check: %w", err) + } + defer tx.Rollback() + + applied, ledgerExists, err := readAppliedMigrations(ctx, tx) + if err != nil { + return err + } + if err := validateDatabaseState(ctx, tx, ledgerExists, applied, migrations); err != nil { + return err + } + if len(applied) < len(migrations) { + return fmt.Errorf( + "%w: schema version %d, this build expects %d; the OpenMessage app applies them when it starts", + ErrMigrationPending, + len(applied), + len(migrations), + ) + } + return nil +} + +// existingStoreDSN is storeDSN with mode=rw: SQLite opens the file for +// reading and writing but refuses to create it. +func existingStoreDSN(path string) string { + dsn, err := url.Parse(storeDSN(path)) + if err != nil { + return storeDSN(path) + } + query := dsn.Query() + query.Set("mode", "rw") + dsn.RawQuery = query.Encode() + return dsn.String() +} diff --git a/internal/storage/sqlite/open_without_migrating_test.go b/internal/storage/sqlite/open_without_migrating_test.go new file mode 100644 index 00000000..2f94e4db --- /dev/null +++ b/internal/storage/sqlite/open_without_migrating_test.go @@ -0,0 +1,195 @@ +package sqlite + +import ( + "context" + "database/sql" + "errors" + "os" + "path/filepath" + "slices" + "strings" + "testing" + "time" +) + +// openStoreAtVersion builds a store migrated through the first version +// migrations only, the shape of a store the app has not upgraded yet. +func openStoreAtVersion(t *testing.T, path string, version int) { + t.Helper() + database, err := sql.Open("sqlite", storeDSN(path)) + if err != nil { + t.Fatalf("sql.Open(): %v", err) + } + defer database.Close() + if err := enableWAL(context.Background(), database); err != nil { + t.Fatalf("enableWAL(): %v", err) + } + if err := runMigrations(context.Background(), database, embeddedMigrations[:version]); err != nil { + t.Fatalf("runMigrations(%d): %v", version, err) + } +} + +func TestOpenWithoutMigratingServesACurrentStore(t *testing.T) { + path := filepath.Join(t.TempDir(), "store.sqlite3") + owner, err := Open(path) + if err != nil { + t.Fatalf("Open(): %v", err) + } + mustRepositoryWrite(t, "UpsertAccount", owner.UpsertAccount(repositoryTestAccount("account-a"))) + ledger := readLedgerRows(t, owner.db) + if err := owner.Close(); err != nil { + t.Fatalf("Close(): %v", err) + } + + client, err := OpenWithoutMigrating(path) + if err != nil { + t.Fatalf("OpenWithoutMigrating(): %v", err) + } + defer client.Close() + if _, err := client.GetAccount("account-a"); err != nil { + t.Fatalf("GetAccount through the client store: %v", err) + } + if got := readLedgerRows(t, client.db); !slices.Equal(got, ledger) { + t.Fatalf("ledger changed:\nbefore: %+v\nafter: %+v", ledger, got) + } +} + +// A newer client meeting a store the app has not upgraded refuses it: no +// migration, no ledger row, no user_version change. (Open would apply the +// pending migrations under the app's daemon.) The fixture has no WAL left to +// checkpoint, so the file also stays byte-identical. +func TestOpenWithoutMigratingRefusesAStoreWithPendingMigrations(t *testing.T) { + path := filepath.Join(t.TempDir(), "store.sqlite3") + behind := len(embeddedMigrations) - 1 + openStoreAtVersion(t, path, behind) + before, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + + store, err := OpenWithoutMigrating(path) + if !errors.Is(err, ErrMigrationPending) { + if store != nil { + store.Close() + } + t.Fatalf("OpenWithoutMigrating() error = %v, want ErrMigrationPending", err) + } + if !strings.Contains(err.Error(), "the OpenMessage app applies them when it starts") { + t.Fatalf("error %q does not say who applies the migrations", err) + } + + database, err := sql.Open("sqlite", storeDSN(path)) + if err != nil { + t.Fatal(err) + } + defer database.Close() + if got := len(readLedgerRows(t, database)); got != behind { + t.Fatalf("ledger rows = %d after the refused open, want %d (nothing applied)", got, behind) + } + assertPragmaInt(t, database, "user_version", behind) + if err := database.Close(); err != nil { + t.Fatal(err) + } + after, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if !slices.Equal(before, after) { + t.Fatal("the refused open changed the database file") + } +} + +func TestOpenWithoutMigratingRefusesNewerAndMissingStores(t *testing.T) { + path := filepath.Join(t.TempDir(), "store.sqlite3") + owner, err := Open(path) + if err != nil { + t.Fatalf("Open(): %v", err) + } + if err := owner.Close(); err != nil { + t.Fatal(err) + } + // A build that knows one migration fewer than the store has applied. + if store, err := openWithoutMigrating(path, embeddedMigrations[:len(embeddedMigrations)-1]); err == nil || + !strings.Contains(err.Error(), "newer than supported") { + if store != nil { + store.Close() + } + t.Fatalf("older build opening a newer store: error = %v, want newer than supported", err) + } + + missing := filepath.Join(t.TempDir(), "missing.sqlite3") + if store, err := OpenWithoutMigrating(missing); err == nil || !errors.Is(err, os.ErrNotExist) { + if store != nil { + store.Close() + } + t.Fatalf("OpenWithoutMigrating(missing) error = %v, want os.ErrNotExist", err) + } + if _, err := os.Stat(missing); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("OpenWithoutMigrating created %s: %v", missing, err) + } + // The connection itself refuses to create the file, so a store removed + // between the existence check and the open is not recreated either. + database, err := sql.Open("sqlite", existingStoreDSN(missing)) + if err != nil { + t.Fatal(err) + } + defer database.Close() + if err := database.Ping(); err == nil { + t.Fatalf("a mode=rw connection to %s opened; want it refused", missing) + } + if _, err := os.Stat(missing); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("the mode=rw connection created %s: %v", missing, err) + } +} + +// The owner may hold SQLite's write reservation for seconds (a migration, a +// large projection). A client open must not queue behind it, and once open the +// client must not hold anything that delays the owner's next write. +func TestOpenWithoutMigratingNeverTakesTheWriteLock(t *testing.T) { + path := filepath.Join(t.TempDir(), "store.sqlite3") + owner, err := Open(path) + if err != nil { + t.Fatalf("Open(): %v", err) + } + defer owner.Close() + ctx := context.Background() + + writer, err := owner.db.BeginTx(ctx, nil) // BEGIN IMMEDIATE via the DSN + if err != nil { + t.Fatalf("owner BeginTx(): %v", err) + } + if _, err := writer.ExecContext(ctx, ` + INSERT INTO accounts (account_id, bridge_key, created_at_ms, updated_at_ms) + VALUES ('held', 'signal', 1, 1) + `); err != nil { + t.Fatalf("owner write: %v", err) + } + start := time.Now() + client, err := OpenWithoutMigrating(path) + elapsed := time.Since(start) + if err != nil { + t.Fatalf("OpenWithoutMigrating() while the owner writes: %v", err) + } + defer client.Close() + if elapsed > time.Second { + t.Fatalf("OpenWithoutMigrating() took %s with the write lock held; it waited for the lock (busy_timeout is %d ms)", elapsed, busyTimeoutMS) + } + if err := writer.Commit(); err != nil { + t.Fatalf("owner Commit(): %v", err) + } + + start = time.Now() + next, err := owner.db.BeginTx(ctx, nil) + if err != nil { + t.Fatalf("owner BeginTx() after the client opened: %v", err) + } + if _, err := next.ExecContext(ctx, `UPDATE accounts SET display_name = 'x' WHERE account_id = 'held'`); err != nil { + t.Fatalf("owner write after the client opened: %v", err) + } + if err := next.Commit(); err != nil { + t.Fatalf("owner Commit(): %v", err) + } + if elapsed := time.Since(start); elapsed > time.Second { + t.Fatalf("owner write took %s with a client store open", elapsed) + } +} diff --git a/internal/storage/sqlite/outbox.go b/internal/storage/sqlite/outbox.go index 9d8e6d71..352dd79f 100644 --- a/internal/storage/sqlite/outbox.go +++ b/internal/storage/sqlite/outbox.go @@ -2042,6 +2042,16 @@ func (r *OutboxRepository) FindStoreFailedDue( return items, nil } +// latestOutboxStateQuery seeks outbox_local_message_idx and reads its newest +// entry; read surfaces run it for every outgoing message they render. +const latestOutboxStateQuery = ` + SELECT state + FROM outbox + WHERE account_id = ? AND local_message_id = ? + ORDER BY created_at_ms DESC, outbox_id DESC + LIMIT 1 +` + // LatestStateForLocalMessage returns the delivery state of the most recent // outbox row for an optimistic outgoing message, so a read surface can show // its send lifecycle. ok is false when the message has no outbox row (a @@ -2052,13 +2062,9 @@ func (r *OutboxRepository) LatestStateForLocalMessage( accountID, localMessageID string, ) (state OutboxState, ok bool, err error) { var value string - scanErr := r.store.db.QueryRowContext(ctx, ` - SELECT state - FROM outbox - WHERE account_id = ? AND local_message_id = ? - ORDER BY created_at_ms DESC, outbox_id DESC - LIMIT 1 - `, accountID, localMessageID).Scan(&value) + scanErr := r.store.db.QueryRowContext( + ctx, latestOutboxStateQuery, accountID, localMessageID, + ).Scan(&value) if errors.Is(scanErr, sql.ErrNoRows) { return "", false, nil } 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/read_path_indexes_migration_test.go b/internal/storage/sqlite/read_path_indexes_migration_test.go new file mode 100644 index 00000000..2960241e --- /dev/null +++ b/internal/storage/sqlite/read_path_indexes_migration_test.go @@ -0,0 +1,172 @@ +package sqlite + +import ( + "context" + "database/sql" + "path/filepath" + "reflect" + "slices" + "testing" +) + +const readPathIndexesChecksum = "a0b486411095e4b1960dba219a0eefbafc243363d450875ba93443fcc0e367f3" + +// readPathIndexesMigration returns the position of the read_path_indexes +// migration, found by name so the test survives a renumbering when parallel +// migrations land first. +func readPathIndexesMigration(t *testing.T) int { + t.Helper() + for i, migration := range embeddedMigrations { + if migration.name == "read_path_indexes" { + return i + } + } + t.Fatal("embedded migrations have no read_path_indexes") + return -1 +} + +func TestReadPathIndexesMigrationAppliesToBlankAndExistingDatabase(t *testing.T) { + position := readPathIndexesMigration(t) + if got := embeddedMigrations[position].checksumSHA256; got != readPathIndexesChecksum { + t.Fatalf("read_path_indexes checksum = %s, want pinned %s (an applied migration must never change)", got, readPathIndexesChecksum) + } + + t.Run("blank", func(t *testing.T) { + store := openRepositoryTestStore(t) + assertReadPathIndexes(t, store.db) + }) + + t.Run("existing store with rows", func(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) + } + if err := enableWAL(context.Background(), database); err != nil { + _ = database.Close() + t.Fatalf("enableWAL(): %v", err) + } + if err := runMigrations(context.Background(), database, embeddedMigrations[:position]); err != nil { + _ = database.Close() + t.Fatalf("runMigrations(before read_path_indexes): %v", err) + } + before := &Store{db: database} + seed := seedReadStoreInto(t, before, 4242) + if len(seed.messages) == 0 { + _ = database.Close() + t.Fatal("seed wrote no messages; the migration would index nothing") + } + ledgerBefore := readLedgerRows(t, database) + rowsBefore := readPathFingerprint(t, database) + if err := database.Close(); err != nil { + t.Fatalf("close pre-migration database: %v", err) + } + + store, err := Open(path) + if err != nil { + t.Fatalf("Open(): %v", err) + } + ledgerAfter := readLedgerRows(t, store.db) + if len(ledgerAfter) != len(embeddedMigrations) { + t.Fatalf("migrated ledger rows = %d, want %d", len(ledgerAfter), len(embeddedMigrations)) + } + if !slices.Equal(ledgerAfter[:position], ledgerBefore) { + t.Fatalf("earlier ledger rows changed:\nbefore: %+v\nafter: %+v", ledgerBefore, ledgerAfter[:position]) + } + assertReadPathIndexes(t, store.db) + if got := readPathFingerprint(t, store.db); !reflect.DeepEqual(got, rowsBefore) { + t.Fatalf("migration changed rows:\nbefore: %v\nafter: %v", rowsBefore, got) + } + var integrity string + if err := store.db.QueryRow(`PRAGMA integrity_check`).Scan(&integrity); err != nil || integrity != "ok" { + t.Fatalf("integrity_check = %q, %v; want ok", integrity, err) + } + if err := store.Close(); err != nil { + t.Fatalf("Close(): %v", err) + } + + reopened, err := Open(path) + if err != nil { + t.Fatalf("reopen: %v", err) + } + t.Cleanup(func() { + if err := reopened.Close(); err != nil { + t.Errorf("Close(): %v", err) + } + }) + if got := readLedgerRows(t, reopened.db); !slices.Equal(got, ledgerAfter) { + t.Fatalf("reopen changed the ledger:\nbefore: %+v\nafter: %+v", ledgerAfter, got) + } + }) +} + +func assertReadPathIndexes(t *testing.T, db *sql.DB) { + t.Helper() + position := readPathIndexesMigration(t) + ledger := readLedgerRow(t, db, position+1) + if ledger.name != "read_path_indexes" || ledger.checksum != readPathIndexesChecksum { + t.Fatalf("ledger row %d = %+v, want read_path_indexes with the pinned checksum", position+1, ledger) + } + for _, index := range []struct { + name string + table string + columns []string + partial bool + }{ + {"messages_account_direction_time_idx", "messages", []string{"account_id", "direction", "occurred_at_ms"}, false}, + {"messages_time_idx", "messages", []string{"occurred_at_ms", "message_id"}, false}, + {"outbox_local_message_idx", "outbox", []string{"local_message_id", "account_id", "created_at_ms", "outbox_id"}, true}, + } { + rows, err := db.Query(`SELECT name FROM pragma_index_info(?) ORDER BY seqno`, index.name) + if err != nil { + t.Fatalf("read %s columns: %v", index.name, err) + } + var columns []string + for rows.Next() { + var column string + if err := rows.Scan(&column); err != nil { + t.Fatalf("scan %s column: %v", index.name, err) + } + columns = append(columns, column) + } + rows.Close() + if !slices.Equal(columns, index.columns) { + t.Fatalf("%s columns = %v, want %v", index.name, columns, index.columns) + } + var table string + var partial bool + if err := db.QueryRow(` + SELECT tbl_name, sql LIKE '%WHERE%' FROM sqlite_schema WHERE type = 'index' AND name = ? + `, index.name).Scan(&table, &partial); err != nil { + t.Fatalf("read %s definition: %v", index.name, err) + } + if table != index.table || partial != index.partial { + t.Fatalf("%s on %s partial=%v, want on %s partial=%v", index.name, table, partial, index.table, index.partial) + } + } +} + +// readPathFingerprint lists every row the new indexes cover, so the test can +// show the migration only adds indexes. +func readPathFingerprint(t *testing.T, db *sql.DB) []string { + t.Helper() + var out []string + for _, query := range []string{ + `SELECT message_id || '|' || account_id || '|' || direction || '|' || occurred_at_ms FROM messages ORDER BY message_id`, + `SELECT outbox_id || '|' || account_id || '|' || ifnull(local_message_id, '') || '|' || created_at_ms || '|' || state FROM outbox ORDER BY outbox_id`, + } { + rows, err := db.Query(query) + if err != nil { + t.Fatalf("fingerprint: %v", err) + } + for rows.Next() { + var row string + if err := rows.Scan(&row); err != nil { + t.Fatalf("scan fingerprint: %v", err) + } + out = append(out, row) + } + rows.Close() + } + return out +} diff --git a/internal/storage/sqlite/read_plans_property_test.go b/internal/storage/sqlite/read_plans_property_test.go new file mode 100644 index 00000000..2fe6d9a4 --- /dev/null +++ b/internal/storage/sqlite/read_plans_property_test.go @@ -0,0 +1,701 @@ +package sqlite + +import ( + "context" + "database/sql" + "errors" + "fmt" + "math/rand" + "reflect" + "slices" + "sort" + "strings" + "testing" + "testing/quick" + "time" +) + +// The rewrites in the 2026-10-09 plan audit change how SQLite reaches rows, +// never which rows: each property below runs the new statement and the +// statement it replaced (kept here verbatim as the reference) over random +// stores and requires identical answers. The stores are small and dense on +// purpose: timestamps collide, phone numbers repeat across accounts, senders +// are NULL, threads are archived and rosters overlap. + +// oldMessagesBeforeCursorQuery and oldMessagesAfterCursorQuery are the OR-form +// keyset cursors the row-value constants replaced. +const ( + oldMessagesBeforeCursorQuery = ` + SELECT ` + messageColumns + ` + FROM messages + WHERE conversation_id = ? + AND (occurred_at_ms < ? OR (occurred_at_ms = ? AND message_id < ?)) + ORDER BY occurred_at_ms DESC, message_id DESC + LIMIT ? + ` + oldMessagesAfterCursorQuery = ` + SELECT ` + messageColumns + ` + FROM messages + WHERE conversation_id = ? + AND (occurred_at_ms > ? OR (occurred_at_ms = ? AND message_id > ?)) + ORDER BY occurred_at_ms ASC, message_id ASC + LIMIT ? + ` + oldConversationPeerIdentitiesQuery = ` + SELECT i.identity_id, i.account_id, i.kind, i.canonical_value, i.raw_value, + i.display_name, i.is_self, i.metadata_json, i.created_at_ms, i.updated_at_ms + FROM conversation_participants p + JOIN identities i + ON i.account_id = p.account_id AND i.identity_id = p.identity_id + WHERE p.account_id = ? AND p.conversation_id = ? + AND p.is_active = 1 AND i.is_self = 0 + ORDER BY i.identity_id + ` + oldDirectConversationBySolePeerQuery = ` + SELECT ` + conversationColumns + ` + FROM conversations c + WHERE c.account_id = ? AND c.kind = 'direct' + AND EXISTS ( + SELECT 1 FROM conversation_participants p + WHERE p.account_id = c.account_id AND p.conversation_id = c.conversation_id + AND p.identity_id = ? AND p.is_active = 1 + ) + AND NOT EXISTS ( + SELECT 1 FROM conversation_participants p2 + JOIN identities i2 + ON i2.account_id = p2.account_id AND i2.identity_id = p2.identity_id + WHERE p2.account_id = c.account_id AND p2.conversation_id = c.conversation_id + 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 + ` + oldGroupConversationsQuery = ` + SELECT ` + conversationColumns + ` + FROM conversations c + WHERE c.account_id = ? AND c.kind = 'group' + ORDER BY c.last_message_at_ms DESC, c.conversation_id + ` + oldMessageHasReadCursorQuery = `SELECT COUNT(*) FROM read_cursors WHERE last_read_message_id = ?` + oldReadCursorDeviceOutsideQuery = ` + SELECT device_id FROM read_cursors + WHERE last_read_message_id = ? AND conversation_id <> ? + ORDER BY device_id + LIMIT 1 + ` +) + +// oldSearchMessagesStatement is SearchMessages' builder before the audit: the +// sender filter joined identities and compared i.canonical_value. +func oldSearchMessagesStatement(query string, filter SearchQuery) (string, []any) { + conditions := []string{"m.body LIKE '%' || ? || '%'"} + args := []any{query} + if filter.AccountID != "" { + conditions = append(conditions, "m.account_id = ?") + args = append(args, filter.AccountID) + } + if filter.ConversationID != "" { + conditions = append(conditions, "m.conversation_id = ?") + args = append(args, filter.ConversationID) + } + if filter.SenderCanonicalValue != "" { + conditions = append(conditions, "i.canonical_value = ?") + args = append(args, filter.SenderCanonicalValue) + } + if filter.SinceMS > 0 { + conditions = append(conditions, "m.occurred_at_ms >= ?") + args = append(args, filter.SinceMS) + } + if filter.UntilMS > 0 { + conditions = append(conditions, "m.occurred_at_ms <= ?") + args = append(args, filter.UntilMS) + } + args = append(args, filter.Limit) + return ` + SELECT ` + prefixedMessageColumns("m") + ` + FROM messages AS m + LEFT JOIN identities AS i + ON i.account_id = m.account_id AND i.identity_id = m.sender_identity_id + WHERE ` + strings.Join(conditions, " AND ") + ` + ORDER BY m.occurred_at_ms DESC, m.message_id DESC + LIMIT ? + `, args +} + +// readPlanStore is a random store plus the probe values a property draws from. +type readPlanStore struct { + Seed int64 +} + +func (readPlanStore) Generate(r *rand.Rand, _ int) reflect.Value { + return reflect.ValueOf(readPlanStore{Seed: r.Int63()}) +} + +type seededReadStore struct { + store *Store + accounts []string + conversations []Conversation + identities []Identity + messages []Message + phones []string + terms []string + outboxLocal []string // local_message_id values that have outbox rows +} + +var readPlanBodies = []string{"hello", "the plan", "zebra crossing", "", "lunch?", "THE END", "hello again"} + +// seedReadStore fills a fresh store from seed. All writes go through SQL so the +// shapes the repositories would refuse (none are needed here) cannot sneak in, +// and every foreign key holds. +func seedReadStore(t *testing.T, seed int64) seededReadStore { + t.Helper() + return seedReadStoreInto(t, openRepositoryTestStore(t), seed) +} + +// seedReadStoreInto writes seedReadStore's rows into an open store. +func seedReadStoreInto(t *testing.T, store *Store, seed int64) seededReadStore { + t.Helper() + r := rand.New(rand.NewSource(seed)) + db := store.db + exec := func(query string, args ...any) { + t.Helper() + if _, err := db.Exec(query, args...); err != nil { + t.Fatalf("seed: %v\n%s %v", err, query, args) + } + } + out := seededReadStore{store: store, terms: []string{"", "hello", "the", "zebra", "absent-term", "?"}} + nAccounts := 1 + r.Intn(3) + for a := 0; a < nAccounts; a++ { + id := fmt.Sprintf("account-%d", a) + out.accounts = append(out.accounts, id) + exec(`INSERT INTO accounts (account_id, bridge_key, created_at_ms, updated_at_ms) VALUES (?, ?, 1, 1)`, + id, []string{"google_messages", "signal_cli", "whatsmeow"}[a]) + exec(`INSERT INTO devices (device_id, account_id, kind, is_current, created_at_ms, updated_at_ms) VALUES (?, ?, 'local_installation', 1, 1, 1)`, + "device-"+id, id) + } + // Phone numbers deliberately repeat across accounts and kinds. + out.phones = []string{"+15550000001", "+15550000002", "+15550000003", "+15550000004"} + identityNo := 0 + for _, account := range out.accounts { + self := Identity{IdentityID: fmt.Sprintf("identity-%d", identityNo), AccountID: account, Kind: "phone", CanonicalValue: "self-" + account, IsSelf: true} + identityNo++ + out.identities = append(out.identities, self) + for k := 0; k < 2+r.Intn(5); k++ { + kind := []string{"phone", "e164", "uuid"}[r.Intn(3)] + value := out.phones[r.Intn(len(out.phones))] + if kind == "uuid" { + value = fmt.Sprintf("uuid-%d", r.Intn(6)) + } + identity := Identity{IdentityID: fmt.Sprintf("identity-%d", identityNo), AccountID: account, Kind: IdentityKind(kind), CanonicalValue: value} + identityNo++ + // (account, kind, canonical) is unique; skip a collision. + if slices.ContainsFunc(out.identities, func(i Identity) bool { + return i.AccountID == account && i.Kind == identity.Kind && i.CanonicalValue == value + }) { + continue + } + out.identities = append(out.identities, identity) + } + } + for _, identity := range out.identities { + exec(`INSERT INTO identities (identity_id, account_id, kind, canonical_value, raw_value, is_self, created_at_ms, updated_at_ms) VALUES (?, ?, ?, ?, ?, ?, 1, 1)`, + identity.IdentityID, identity.AccountID, identity.Kind, identity.CanonicalValue, identity.CanonicalValue, identity.IsSelf) + } + accountIdentities := func(account string) []Identity { + var ids []Identity + for _, identity := range out.identities { + if identity.AccountID == account { + ids = append(ids, identity) + } + } + return ids + } + nConversations := 1 + r.Intn(10) + for c := 0; c < nConversations; c++ { + account := out.accounts[r.Intn(len(out.accounts))] + conversation := Conversation{ + ConversationID: fmt.Sprintf("conversation-%02d", c), + AccountID: account, + Kind: []ConversationKind{ConversationKindDirect, ConversationKindGroup}[r.Intn(2)], + LastMessageAtMS: int64(r.Intn(6)) * 10, // ties on purpose + NotificationMode: NotificationModeAll, + } + var archived any + if r.Intn(4) == 0 { + archivedAt := int64(r.Intn(3)) + archived = archivedAt + conversation.ArchivedAtMS = &archivedAt + } + out.conversations = append(out.conversations, conversation) + exec(`INSERT INTO conversations (conversation_id, account_id, remote_conversation_id, kind, archived_at_ms, last_message_at_ms, created_at_ms, updated_at_ms) VALUES (?, ?, ?, ?, ?, ?, 1, 1)`, + conversation.ConversationID, account, "remote-"+conversation.ConversationID, conversation.Kind, archived, conversation.LastMessageAtMS) + for _, identity := range accountIdentities(account) { + if r.Intn(2) == 0 { + continue + } + exec(`INSERT INTO conversation_participants (account_id, conversation_id, identity_id, is_active) VALUES (?, ?, ?, ?)`, + account, conversation.ConversationID, identity.IdentityID, r.Intn(4) != 0) + } + } + nMessages := r.Intn(60) + for m := 0; m < nMessages; m++ { + conversation := out.conversations[r.Intn(len(out.conversations))] + var sender *string + if candidates := accountIdentities(conversation.AccountID); r.Intn(4) != 0 { + id := candidates[r.Intn(len(candidates))].IdentityID + sender = &id + } + message := Message{ + // IDs are not in time order, so ties are broken by ID, not insertion. + MessageID: fmt.Sprintf("message-%02d", r.Intn(1000)), + ConversationID: conversation.ConversationID, + AccountID: conversation.AccountID, + RemoteMessageID: fmt.Sprintf("remote-%d", m), + SenderIdentityID: sender, + Direction: []MessageDirection{MessageDirectionIncoming, MessageDirectionOutgoing}[r.Intn(2)], + Body: readPlanBodies[r.Intn(len(readPlanBodies))], + OccurredAtMS: 1 + int64(r.Intn(8)), // dense: many ties + } + if slices.ContainsFunc(out.messages, func(other Message) bool { return other.MessageID == message.MessageID }) { + continue + } + out.messages = append(out.messages, message) + exec(`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 (?, ?, ?, ?, ?, ?, ?, ?, 1, 1)`, + message.MessageID, message.ConversationID, message.AccountID, message.RemoteMessageID, message.SenderIdentityID, message.Direction, message.Body, message.OccurredAtMS) + } + // Read cursors point at some messages; outbox rows name some as local IDs. + for _, message := range out.messages { + if r.Intn(5) == 0 { + exec(`INSERT OR IGNORE INTO read_cursors (account_id, device_id, conversation_id, last_read_message_id, last_read_at_ms, updated_at_ms) VALUES (?, ?, ?, ?, 1, 1)`, + message.AccountID, "device-"+message.AccountID, message.ConversationID, message.MessageID) + } + for k := 0; k < r.Intn(3); k++ { + outboxID := fmt.Sprintf("outbox-%s-%d", message.MessageID, k) + exec(`INSERT INTO outbox (outbox_id, account_id, conversation_id, kind, idempotency_key, payload_hash, operation, state, local_message_id, transport_request_id, scheduled_for_ms, created_at_ms, updated_at_ms) VALUES (?, ?, ?, 'text', ?, 'h', 'send', ?, ?, ?, 1, ?, ?)`, + outboxID, message.AccountID, message.ConversationID, outboxID, []string{"confirmed", "rejected", "canceled"}[r.Intn(3)], message.MessageID, outboxID, int64(1+r.Intn(3)), int64(10)) + out.outboxLocal = append(out.outboxLocal, message.MessageID) + } + } + return out +} + +// queryMessages returns every column of every row, so a differential check +// compares whole rows, not just which IDs came back. +func queryMessages(t *testing.T, db *sql.DB, query string, args ...any) []Message { + t.Helper() + rows, err := db.Query(query, args...) + if err != nil { + t.Fatalf("query: %v\n%s", err, query) + } + messages, err := collectRows(rows, scanMessage) + if err != nil { + t.Fatalf("scan: %v", err) + } + return messages +} + +func queryConversations(t *testing.T, db *sql.DB, query string, args ...any) []Conversation { + t.Helper() + rows, err := db.Query(query, args...) + if err != nil { + t.Fatalf("query: %v\n%s", err, query) + } + conversations, err := collectRows(rows, scanConversation) + if err != nil { + t.Fatalf("scan: %v", err) + } + return conversations +} + +func queryConversationIDs(t *testing.T, db *sql.DB, query string, args ...any) []string { + t.Helper() + conversations := queryConversations(t, db, query, args...) + ids := make([]string, 0, len(conversations)) + for _, conversation := range conversations { + ids = append(ids, conversation.ConversationID) + } + return ids +} + +var readPlanQuickConfig = &quick.Config{MaxCount: 30} + +// Invariant: the row-value keyset cursor selects exactly the rows of the +// OR-form cursor, in the same order, for every cursor and limit; and paging a +// conversation with it visits every message once, newest first. +func TestKeysetCursorMatchesORFormProperty(t *testing.T) { + property := func(c readPlanStore) bool { + s := seedReadStore(t, c.Seed) + r := rand.New(rand.NewSource(c.Seed ^ 0x5eed)) + ids := []string{"", "message-00", "message-500", "message-999", "zzz"} + for _, message := range s.messages { + ids = append(ids, message.MessageID) + } + for range 40 { + conversation := s.conversations[r.Intn(len(s.conversations))].ConversationID + ms := int64(r.Intn(11)) - 1 + id := ids[r.Intn(len(ids))] + limit := 1 + r.Intn(8) + for _, pair := range [][2]string{ + {messagesBeforeCursorQuery, oldMessagesBeforeCursorQuery}, + {messagesAfterCursorQuery, oldMessagesAfterCursorQuery}, + } { + got := queryMessages(t, s.store.db, pair[0], conversation, ms, id, limit) + want := queryMessages(t, s.store.db, pair[1], conversation, ms, ms, id, limit) + if !reflect.DeepEqual(got, want) { + t.Errorf("seed %d: cursor (%d, %q) limit %d in %s = %v, OR form = %v", c.Seed, ms, id, limit, conversation, messageIDs(got), messageIDs(want)) + return false + } + } + } + repository := mustMessageRepository(t, s.store, 100) + for _, conversation := range s.conversations { + var want []Message + for _, message := range s.messages { + if message.ConversationID == conversation.ConversationID { + want = append(want, message) + } + } + sort.Slice(want, func(i, j int) bool { + if want[i].OccurredAtMS != want[j].OccurredAtMS { + return want[i].OccurredAtMS > want[j].OccurredAtMS + } + return want[i].MessageID > want[j].MessageID + }) + var walked []string + var beforeMS int64 + var beforeID string + for page := 0; page <= len(want)+1; page++ { + rows, err := repository.ListMessagesByConversation(context.Background(), conversation.ConversationID, beforeMS, beforeID, 3) + if err != nil { + t.Fatalf("ListMessagesByConversation: %v", err) + } + for _, row := range rows { + walked = append(walked, row.MessageID) + } + if len(rows) < 3 { + break + } + beforeMS, beforeID = rows[len(rows)-1].OccurredAtMS, rows[len(rows)-1].MessageID + } + wantIDs := make([]string, 0, len(want)) + for _, message := range want { + wantIDs = append(wantIDs, message.MessageID) + } + if !slices.Equal(walked, wantIDs) { + t.Errorf("seed %d: walking %s in pages of 3 = %v, want %v", c.Seed, conversation.ConversationID, walked, wantIDs) + return false + } + } + return true + } + if err := quick.Check(property, readPlanQuickConfig); err != nil { + t.Fatal(err) + } +} + +// Invariant: for every filter combination, the audited search statement +// returns the same messages, in the same order, as the statement it replaced. +func TestSearchStatementMatchesPreviousStatementProperty(t *testing.T) { + property := func(c readPlanStore) bool { + s := seedReadStore(t, c.Seed) + r := rand.New(rand.NewSource(c.Seed ^ 0x5ea7c4)) + pick := func(values []string) string { return values[r.Intn(len(values))] } + for range 60 { + filter := SearchQuery{Limit: 1 + r.Intn(12)} + if r.Intn(4) == 0 { + filter.AccountID = pick(append(slices.Clone(s.accounts), "account-none")) + } + if r.Intn(3) == 0 { + filter.ConversationID = s.conversations[r.Intn(len(s.conversations))].ConversationID + } + if r.Intn(2) == 0 { + filter.SenderCanonicalValue = pick(append(slices.Clone(s.phones), "uuid-1", "self-account-0", "+19999999999")) + } + if r.Intn(3) == 0 { + filter.SinceMS = int64(r.Intn(9)) + } + if r.Intn(3) == 0 { + filter.UntilMS = int64(r.Intn(9)) + } + term := pick(s.terms) + newSQL, newArgs := searchMessagesStatement(term, filter) + oldSQL, oldArgs := oldSearchMessagesStatement(term, filter) + got := queryMessages(t, s.store.db, newSQL, newArgs...) + want := queryMessages(t, s.store.db, oldSQL, oldArgs...) + if !reflect.DeepEqual(got, want) { + t.Errorf("seed %d: search %q %+v = %v, previous statement = %v", c.Seed, term, filter, messageIDs(got), messageIDs(want)) + return false + } + } + return true + } + if err := quick.Check(property, readPlanQuickConfig); err != nil { + t.Fatal(err) + } +} + +// Invariant: the rebind lookups return what the replaced statements and the +// replaced group scan returned, for every account, identity and peer set. +func TestRebindLookupsMatchPreviousQueriesProperty(t *testing.T) { + property := func(c readPlanStore) bool { + s := seedReadStore(t, c.Seed) + r := rand.New(rand.NewSource(c.Seed ^ 0x2eb1d)) + db := s.store.db + identityIDs := []string{"identity-missing"} + for _, identity := range s.identities { + identityIDs = append(identityIDs, identity.IdentityID) + } + for _, account := range s.accounts { + for _, conversation := range s.conversations { + got, err := s.store.ListConversationPeerIdentities(account, conversation.ConversationID) + if err != nil { + t.Fatalf("ListConversationPeerIdentities: %v", err) + } + rows, err := db.Query(oldConversationPeerIdentitiesQuery, account, conversation.ConversationID) + if err != nil { + t.Fatalf("old peers: %v", err) + } + want := []Identity{} + for rows.Next() { + var identity Identity + if err := rows.Scan(&identity.IdentityID, &identity.AccountID, &identity.Kind, &identity.CanonicalValue, + &identity.RawValue, &identity.DisplayName, &identity.IsSelf, &identity.MetadataJSON, + &identity.CreatedAtMS, &identity.UpdatedAtMS); err != nil { + t.Fatalf("scan old peers: %v", err) + } + want = append(want, identity) + } + rows.Close() + if !reflect.DeepEqual(append([]Identity{}, got...), want) { + t.Errorf("seed %d: peers of %s/%s = %+v, previous = %+v", c.Seed, account, conversation.ConversationID, got, want) + return false + } + } + for _, identityID := range identityIDs { + got, gotErr := s.store.FindDirectConversationBySolePeer(account, identityID) + want := queryConversations(t, db, oldDirectConversationBySolePeerQuery, account, identityID, identityID) + switch { + case len(want) == 0 && !errorsIsNotFound(gotErr): + t.Errorf("seed %d: sole peer %s/%s = %v, %v; previous found none", c.Seed, account, identityID, got.ConversationID, gotErr) + return false + case len(want) == 1 && (gotErr != nil || !reflect.DeepEqual(got, want[0])): + t.Errorf("seed %d: sole peer %s/%s = %+v, %v; previous = %+v", c.Seed, account, identityID, got, gotErr, want[0]) + return false + } + } + // Peer sets: every active roster in the store, plus random sets. + sets := [][]string{} + for _, conversation := range s.conversations { + peers, _ := s.store.ListConversationPeerIdentities(account, conversation.ConversationID) + var set []string + for _, peer := range peers { + set = append(set, peer.IdentityID) + } + if len(set) > 0 { + r.Shuffle(len(set), func(i, j int) { set[i], set[j] = set[j], set[i] }) + sets = append(sets, set) + } + } + for range 6 { + var set []string + for range 1 + r.Intn(3) { + set = append(set, identityIDs[r.Intn(len(identityIDs))]) + } + sets = append(sets, set) + } + for _, set := range sets { + got, gotErr := s.store.FindGroupConversationByPeerSet(account, set) + want, wantFound := previousGroupByPeerSet(t, db, account, set) + if wantFound != (gotErr == nil) || (wantFound && !reflect.DeepEqual(got, want)) { + t.Errorf("seed %d: group with peers %v in %s = %+v, %v; previous = %+v, found %v", c.Seed, set, account, got, gotErr, want, wantFound) + return false + } + } + } + return true + } + if err := quick.Check(property, readPlanQuickConfig); err != nil { + t.Fatal(err) + } +} + +func errorsIsNotFound(err error) bool { + return errors.Is(err, ErrNotFound) +} + +// previousGroupByPeerSet is FindGroupConversationByPeerSet as it was: every +// group of the account newest first, each roster compared with the wanted set. +func previousGroupByPeerSet(t *testing.T, db *sql.DB, account string, identityIDs []string) (Conversation, bool) { + t.Helper() + want := map[string]struct{}{} + for _, id := range identityIDs { + want[id] = struct{}{} + } + for _, group := range queryConversations(t, db, oldGroupConversationsQuery, account) { + rows, err := db.Query(oldConversationPeerIdentitiesQuery, account, group.ConversationID) + if err != nil { + t.Fatalf("old roster: %v", err) + } + var peers []string + for rows.Next() { + var identityID, skip string + var isSelf bool + var created, updated int64 + if err := rows.Scan(&identityID, &skip, &skip, &skip, &skip, &skip, &isSelf, &skip, &created, &updated); err != nil { + t.Fatalf("scan old roster: %v", err) + } + peers = append(peers, identityID) + } + rows.Close() + if len(peers) != len(want) { + continue + } + match := true + for _, peer := range peers { + if _, ok := want[peer]; !ok { + match = false + break + } + } + if match { + return group, true + } + } + return Conversation{}, false +} + +// Invariant: the one-statement recency list equals sorting every conversation +// of every account (archived ones included) and keeping the first limit. +func TestConversationsByRecencyMatchesFullSortProperty(t *testing.T) { + property := func(c readPlanStore) bool { + s := seedReadStore(t, c.Seed) + // Every conversation row, read whole, sorted in Go: the reference + // does not share the statement under test. + all := queryConversations(t, s.store.db, `SELECT `+conversationColumns+` FROM conversations NOT INDEXED`) + if len(all) != len(s.conversations) { + t.Fatalf("seed %d: read %d conversations, seeded %d", c.Seed, len(all), len(s.conversations)) + } + sort.Slice(all, func(i, j int) bool { + if all[i].LastMessageAtMS != all[j].LastMessageAtMS { + return all[i].LastMessageAtMS > all[j].LastMessageAtMS + } + return all[i].ConversationID < all[j].ConversationID + }) + for limit := 0; limit <= len(all)+2; limit++ { + got, err := s.store.ListConversationsByRecencyAllAccounts(limit) + if err != nil { + t.Fatalf("ListConversationsByRecencyAllAccounts(%d): %v", limit, err) + } + want := all[:min(limit, len(all))] + if !reflect.DeepEqual(got, want) { + t.Errorf("seed %d: limit %d returned %v, want %v", c.Seed, limit, conversationIDs(got), conversationIDs(want)) + return false + } + } + return true + } + if err := quick.Check(property, readPlanQuickConfig); err != nil { + t.Fatal(err) + } +} + +// Invariant: the per-account aggregates equal the same aggregates computed +// over the seeded rows: latest is the max occurred_at_ms (0 when none), latest +// incoming is the max over incoming rows, and the counts are row counts. +// MessageHasReadCursor agrees with counting cursors directly, and the outbox +// state lookup returns the newest row by (created_at_ms, outbox_id). +func TestAccountAggregatesAndLookupsMatchReferenceProperty(t *testing.T) { + property := func(c readPlanStore) bool { + s := seedReadStore(t, c.Seed) + for _, account := range append(slices.Clone(s.accounts), "account-none") { + var wantLatest, wantIncoming, wantMessages, wantConversations int64 + for _, message := range s.messages { + if message.AccountID != account { + continue + } + wantMessages++ + wantLatest = max(wantLatest, message.OccurredAtMS) + if message.Direction == MessageDirectionIncoming { + wantIncoming = max(wantIncoming, message.OccurredAtMS) + } + } + for _, conversation := range s.conversations { + if conversation.AccountID == account { + wantConversations++ + } + } + latest, incoming, err := s.store.LatestMessageTimes(account) + if err != nil || latest != wantLatest || incoming != wantIncoming { + t.Errorf("seed %d: LatestMessageTimes(%s) = %d, %d, %v; want %d, %d", c.Seed, account, latest, incoming, err, wantLatest, wantIncoming) + return false + } + messages, err := s.store.CountMessages(account) + if err != nil || messages != wantMessages { + t.Errorf("seed %d: CountMessages(%s) = %d, %v; want %d", c.Seed, account, messages, err, wantMessages) + return false + } + conversations, err := s.store.CountConversations(account) + if err != nil || conversations != wantConversations { + t.Errorf("seed %d: CountConversations(%s) = %d, %v; want %d", c.Seed, account, conversations, err, wantConversations) + return false + } + } + for _, id := range append([]string{"message-missing"}, messageIDs(s.messages)...) { + got, err := s.store.MessageHasReadCursor(id) + var count int64 + if scanErr := s.store.db.QueryRow(oldMessageHasReadCursorQuery, id).Scan(&count); scanErr != nil { + t.Fatalf("old cursor count: %v", scanErr) + } + if err != nil || got != (count > 0) { + t.Errorf("seed %d: MessageHasReadCursor(%s) = %v, %v; cursor count %d", c.Seed, id, got, err, count) + return false + } + for _, conversation := range append(conversationIDs(s.conversations), "conversation-none") { + var gotDevice, wantDevice string + gotErr := s.store.db.QueryRow(readCursorDeviceOutsideQuery, id, conversation, id).Scan(&gotDevice) + wantErr := s.store.db.QueryRow(oldReadCursorDeviceOutsideQuery, id, conversation).Scan(&wantDevice) + if (gotErr == nil) != (wantErr == nil) || gotDevice != wantDevice { + t.Errorf("seed %d: cursor device outside %s for %s = %q (%v), previous = %q (%v)", c.Seed, conversation, id, gotDevice, gotErr, wantDevice, wantErr) + return false + } + } + } + outbox, err := NewOutboxRepository(s.store, func() time.Time { return time.UnixMilli(100) }) + if err != nil { + t.Fatalf("NewOutboxRepository: %v", err) + } + for _, message := range s.messages { + got, ok, err := outbox.LatestStateForLocalMessage(context.Background(), message.AccountID, message.MessageID) + var want string + wantErr := s.store.db.QueryRow(` + SELECT state FROM outbox NOT INDEXED + WHERE account_id = ? AND local_message_id = ? + ORDER BY created_at_ms DESC, outbox_id DESC LIMIT 1 + `, message.AccountID, message.MessageID).Scan(&want) + if err != nil || ok != (wantErr == nil) || (ok && string(got) != want) { + t.Errorf("seed %d: LatestStateForLocalMessage(%s) = %q, %v, %v; want %q (%v)", c.Seed, message.MessageID, got, ok, err, want, wantErr) + return false + } + } + return true + } + if err := quick.Check(property, readPlanQuickConfig); err != nil { + t.Fatal(err) + } +} + +func messageIDs(messages []Message) []string { + ids := make([]string, 0, len(messages)) + for _, message := range messages { + ids = append(ids, message.MessageID) + } + return ids +} + +func conversationIDs(conversations []Conversation) []string { + ids := make([]string, 0, len(conversations)) + for _, conversation := range conversations { + ids = append(ids, conversation.ConversationID) + } + return ids +} diff --git a/internal/storage/sqlite/read_plans_test.go b/internal/storage/sqlite/read_plans_test.go new file mode 100644 index 00000000..852c9f18 --- /dev/null +++ b/internal/storage/sqlite/read_plans_test.go @@ -0,0 +1,304 @@ +package sqlite + +import ( + "database/sql" + "slices" + "strings" + "testing" +) + +// explainPlanLines returns the EXPLAIN QUERY PLAN detail lines for query with +// args bound, in plan order. Bound values matter: a partial index is usable +// only when the query's WHERE implies the index's WHERE. +func explainPlanLines(t *testing.T, db *sql.DB, query string, args ...any) []string { + t.Helper() + rows, err := db.Query("EXPLAIN QUERY PLAN "+query, args...) + if err != nil { + t.Fatalf("EXPLAIN QUERY PLAN: %v\n%s", err, query) + } + defer rows.Close() + var lines []string + for rows.Next() { + var id, parent, notUsed int + var detail string + if err := rows.Scan(&id, &parent, ¬Used, &detail); err != nil { + t.Fatalf("scan plan row: %v", err) + } + lines = append(lines, detail) + } + if err := rows.Err(); err != nil { + t.Fatalf("read plan rows: %v", err) + } + return lines +} + +// planScans reports the plan lines that read a whole table or a whole index: +// "SCAN t" or "SCAN t USING [COVERING] INDEX i". Scans of a subquery or CTE +// result ("SCAN (subquery-1)", "SCAN CONSTANT ROW") read rows already bounded +// by their own plan lines and are not counted. +func planScans(plan []string, table string) []string { + var scans []string + for _, line := range plan { + if line == "SCAN "+table || strings.HasPrefix(line, "SCAN "+table+" ") { + scans = append(scans, line) + } + } + return scans +} + +// TestReadQueryPlansStayOffWholeTableScans pins the plans the 2026-10-09 audit +// moved off whole-table reads. The store never runs ANALYZE, so these are the +// default-statistics plans that production uses; an empty store plans exactly +// like a large one without sqlite_stat1. +func TestReadQueryPlansStayOffWholeTableScans(t *testing.T) { + store := openRepositoryTestStore(t) + db := store.db + + type planCase struct { + name string + query string + args []any + // want lists lines the plan must contain; the scanned tables must not + // be read whole. + want []string + notScan []string + wantExact bool + } + const ( + conv = "conversation-a" + account = "account-a" + ) + search := func(term string, filter SearchQuery) (string, []any) { + filter.Limit = max(filter.Limit, 1) + return searchMessagesStatement(term, filter) + } + senderSQL, senderArgs := search("", SearchQuery{SenderCanonicalValue: "+15550000001"}) + senderWindowSQL, senderWindowArgs := search("x", SearchQuery{SenderCanonicalValue: "+15550000001", SinceMS: 1, UntilMS: 2}) + listSQL, listArgs := search("", SearchQuery{}) + accountWindowSQL, accountWindowArgs := search("", SearchQuery{AccountID: account, SinceMS: 1, UntilMS: 2}) + accountListSQL, accountListArgs := search("", SearchQuery{AccountID: account}) + listWindowSQL, listWindowArgs := search("", SearchQuery{SinceMS: 1, UntilMS: 2}) + conversationSQL, conversationArgs := search("x", SearchQuery{ConversationID: conv}) + + cases := []planCase{ + { + name: "page before a keyset cursor", + query: messagesBeforeCursorQuery, + args: []any{conv, int64(10), "m", 50}, + want: []string{"SEARCH messages USING INDEX messages_conversation_time_idx (conversation_id=? AND (occurred_at_ms,message_id)<(?,?))"}, + wantExact: true, + }, + { + name: "page after a keyset cursor", + query: messagesAfterCursorQuery, + args: []any{conv, int64(10), "m", 50}, + want: []string{"SEARCH messages USING INDEX messages_conversation_time_idx (conversation_id=? AND (occurred_at_ms,message_id)>(?,?))"}, + wantExact: true, + }, + { + name: "search by sender seeks the sender index per identity", + query: senderSQL, + args: senderArgs, + want: []string{"SEARCH m USING INDEX messages_sender_time_idx (sender_identity_id=?)"}, + notScan: []string{"m", "messages"}, + }, + { + name: "search by sender inside a window", + query: senderWindowSQL, + args: senderWindowArgs, + want: []string{"SEARCH m USING INDEX messages_sender_time_idx (sender_identity_id=? AND occurred_at_ms>? AND occurred_at_ms? AND occurred_at_ms? AND occurred_at_ms?)", + }, + notScan: []string{"conversations"}, + }, + + { + name: "message count stays inside one account's index range", + query: countMessagesQuery, + args: []any{account}, + want: []string{"SEARCH messages USING COVERING INDEX messages_account_direction_time_idx (account_id=?)"}, + notScan: []string{"messages"}, + }, + { + name: "conversation count stays inside one account's index range", + query: countConversationsQuery, + args: []any{account}, + want: []string{"SEARCH conversations USING COVERING INDEX sqlite_autoindex_conversations_3 (account_id=?)"}, + notScan: []string{"conversations"}, + }, + { + name: "outbox state for a rendered outgoing message", + query: latestOutboxStateQuery, + args: []any{account, "message-a"}, + want: []string{"SEARCH outbox USING INDEX outbox_local_message_idx (local_message_id=? AND account_id=?)"}, + wantExact: true, + }, + { + name: "open-send check for a repair row seeks the local message", + query: messageHasOpenSendQuery, + args: []any{"message-a"}, + want: []string{"SEARCH outbox USING INDEX outbox_local_message_idx (local_message_id=?)"}, + notScan: []string{"outbox"}, + }, + { + name: "open send in another conversation seeks the local message", + query: openSendElsewhereQuery, + args: []any{"message-a", conv}, + want: []string{"SEARCH outbox USING INDEX outbox_local_message_idx (local_message_id=?)"}, + notScan: []string{"outbox"}, + }, + { + name: "cursor outside the conversation seeks the message's conversation", + query: readCursorDeviceOutsideQuery, + args: []any{"message-a", conv, "message-a"}, + want: []string{"SEARCH read_cursors USING INDEX read_cursors_conversation_idx (conversation_id=?)"}, + notScan: []string{"read_cursors", "messages"}, + }, + { + name: "read-cursor check seeks the message's conversation", + query: messageHasReadCursorQuery, + args: []any{"message-a", "message-a"}, + want: []string{"SEARCH read_cursors USING INDEX read_cursors_conversation_idx (conversation_id=?)"}, + notScan: []string{"read_cursors", "messages"}, + }, + } + t.Run("content duplicate check seeks the occurrence millisecond", func(t *testing.T) { + plan := explainPlanLines(t, db, messageContentDuplicateQuery, + account, conv, int64(10), "incoming", "body", nil, "remote-a") + if scans := planScans(plan, "messages"); len(scans) > 0 || len(plan) == 0 || + !strings.HasPrefix(plan[0], "SEARCH messages USING INDEX ") || + !strings.Contains(plan[0], "occurred_at_ms=?") { + t.Fatalf("plan = %q, want a SEARCH bounded by occurred_at_ms=?", plan) + } + }) + t.Run("newest message times seek both directions", func(t *testing.T) { + plan := explainPlanLines(t, db, latestMessageTimesQuery, account, account) + seek := "SEARCH messages USING COVERING INDEX messages_account_direction_time_idx (account_id=? AND direction=?)" + seeks := 0 + for _, line := range plan { + if line == seek { + seeks++ + } + } + // One seek per direction. Dropping either subquery's direction term + // keeps the MAX right but walks the account's whole range: still a + // SEARCH, so count the seeks rather than look for a SCAN. + if seeks != 2 || len(planScans(plan, "messages")) > 0 { + t.Fatalf("plan = %q, want exactly two %q lines", plan, seek) + } + }) + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + plan := explainPlanLines(t, db, tc.query, tc.args...) + for _, want := range tc.want { + if !slices.Contains(plan, want) { + t.Fatalf("plan = %q\nwant a line %q", plan, want) + } + } + if tc.wantExact && len(plan) != len(tc.want) { + t.Fatalf("plan = %q\nwant only %q (no scan, no sort)", plan, tc.want) + } + for _, table := range tc.notScan { + if scans := planScans(plan, table); len(scans) > 0 { + t.Fatalf("plan = %q\nreads %s whole: %q", plan, table, scans) + } + } + }) + } +} + +// TestUnscopedSubstringSearchKeepsSequentialScan pins the one deliberate scan: +// a substring search over every conversation reads messages once in rowid order +// and keeps the newest rows in a top-N sort. With messages_time_idx available +// the planner would otherwise walk that index and fetch each row out of order, +// which for a rare or absent term costs several times the scan (3.4 s against +// 0.45 s on a 2.2M-message store). Full-text search is the fix for this one. +func TestUnscopedSubstringSearchKeepsSequentialScan(t *testing.T) { + store := openRepositoryTestStore(t) + for _, filter := range []SearchQuery{ + {Limit: 30}, + {SinceMS: 1, UntilMS: 2, Limit: 30}, + {AccountID: "account-a", Limit: 30}, + } { + query, args := searchMessagesStatement("term", filter) + plan := explainPlanLines(t, store.db, query, args...) + want := []string{"SCAN m", "USE TEMP B-TREE FOR ORDER BY"} + if !slices.Equal(plan, want) { + t.Fatalf("filter %+v: plan = %q, want %q", filter, plan, want) + } + } +} diff --git a/internal/storage/sqlite/rebind.go b/internal/storage/sqlite/rebind.go index 3cea6bf4..6efc9f77 100644 --- a/internal/storage/sqlite/rebind.go +++ b/internal/storage/sqlite/rebind.go @@ -21,22 +21,64 @@ func displacedRemoteID(remoteID, conversationID string) string { return DisplacedRemoteIDPrefix + remoteID + ":" + conversationID } -// ListConversationPeerIdentities returns the active non-self participant -// identities of a conversation — a direct thread's peer, or a group's members. -func (s *Store) ListConversationPeerIdentities( - accountID string, - conversationID string, -) ([]Identity, error) { - rows, err := s.db.QueryContext(context.Background(), ` +// The rebind lookups below start from conversation_participants: a +// conversation's roster through its primary key, a peer's conversations through +// conversation_participants_identity_idx. The store never runs ANALYZE, and +// without statistics the planner otherwise starts from the account's side +// instead (every identity, or every conversation, of the account) and probes +// participants once per row, which grows with the account rather than with the +// roster. +const ( + // CROSS JOIN pins participants as the outer loop. + conversationPeerIdentitiesQuery = ` SELECT i.identity_id, i.account_id, i.kind, i.canonical_value, i.raw_value, i.display_name, i.is_self, i.metadata_json, i.created_at_ms, i.updated_at_ms FROM conversation_participants p - JOIN identities i + CROSS JOIN identities i ON i.account_id = p.account_id AND i.identity_id = p.identity_id WHERE p.account_id = ? AND p.conversation_id = ? AND p.is_active = 1 AND i.is_self = 0 ORDER BY i.identity_id - `, accountID, conversationID) + ` + directConversationBySolePeerQuery = ` + SELECT ` + conversationColumns + ` + FROM conversations c + WHERE c.account_id = ? AND c.kind = 'direct' + AND c.conversation_id IN ( + SELECT p.conversation_id FROM conversation_participants p + WHERE p.identity_id = ? AND p.account_id = ? AND p.is_active = 1 + ) + AND NOT EXISTS ( + SELECT 1 FROM conversation_participants p2 + JOIN identities i2 + ON i2.account_id = p2.account_id AND i2.identity_id = p2.identity_id + WHERE p2.account_id = c.account_id AND p2.conversation_id = c.conversation_id + 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 + ` + groupConversationsWithMemberQuery = ` + SELECT ` + conversationColumns + ` + FROM conversations c + WHERE c.account_id = ? AND c.kind = 'group' + AND c.conversation_id IN ( + SELECT p.conversation_id FROM conversation_participants p + WHERE p.identity_id = ? AND p.account_id = ? AND p.is_active = 1 + ) + ORDER BY c.last_message_at_ms DESC, c.conversation_id + ` +) + +// ListConversationPeerIdentities returns the active non-self participant +// identities of a conversation — a direct thread's peer, or a group's members. +func (s *Store) ListConversationPeerIdentities( + accountID string, + conversationID string, +) ([]Identity, error) { + rows, err := s.db.QueryContext( + context.Background(), conversationPeerIdentitiesQuery, accountID, conversationID, + ) if err != nil { return nil, fmt.Errorf("list peer identities for conversation %q: %w", conversationID, err) } @@ -71,25 +113,10 @@ func (s *Store) FindDirectConversationBySolePeer( accountID string, identityID string, ) (Conversation, error) { - conversation, err := scanConversation(s.db.QueryRowContext(context.Background(), ` - SELECT `+conversationColumns+` - FROM conversations c - WHERE c.account_id = ? AND c.kind = 'direct' - AND EXISTS ( - SELECT 1 FROM conversation_participants p - WHERE p.account_id = c.account_id AND p.conversation_id = c.conversation_id - AND p.identity_id = ? AND p.is_active = 1 - ) - AND NOT EXISTS ( - SELECT 1 FROM conversation_participants p2 - JOIN identities i2 - ON i2.account_id = p2.account_id AND i2.identity_id = p2.identity_id - WHERE p2.account_id = c.account_id AND p2.conversation_id = c.conversation_id - 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)) + conversation, err := scanConversation(s.db.QueryRowContext( + context.Background(), directConversationBySolePeerQuery, + accountID, identityID, accountID, identityID, + )) if errors.Is(err, sql.ErrNoRows) { return Conversation{}, fmt.Errorf( "direct conversation with sole peer %q: %w", @@ -122,12 +149,13 @@ func (s *Store) FindGroupConversationByPeerSet( want[id] = struct{}{} } - rows, err := s.db.QueryContext(context.Background(), ` - SELECT `+conversationColumns+` - FROM conversations c - WHERE c.account_id = ? AND c.kind = 'group' - ORDER BY c.last_message_at_ms DESC, c.conversation_id - `, accountID) + // A match's active non-self participants are exactly the wanted set, so it + // has every wanted identity as an active participant. Only groups holding + // the first one can match; the roster comparison below decides among them. + rows, err := s.db.QueryContext( + context.Background(), groupConversationsWithMemberQuery, + accountID, identityIDs[0], accountID, + ) if err != nil { return Conversation{}, fmt.Errorf("list group conversations: %w", err) } diff --git a/internal/storage/sqlite/repair_support.go b/internal/storage/sqlite/repair_support.go index 3a607c5c..77a61af5 100644 --- a/internal/storage/sqlite/repair_support.go +++ b/internal/storage/sqlite/repair_support.go @@ -62,14 +62,19 @@ func (s *Store) ListInboundSenderIdentityIDsBefore(conversationID string, sinceM // MessageHasReadCursor reports whether any read cursor points at the message. // Such a row cannot change conversation without breaking the cursor's // composite foreign key. +// +// read_cursors has no index on last_read_message_id, but that same composite +// key puts any cursor naming the message in the message's own conversation, so +// the lookup seeks read_cursors_conversation_idx (a few rows, one per device) +// instead of scanning every cursor once per planned repair row. func (s *Store) MessageHasReadCursor(messageID string) (bool, error) { - var count int64 - if err := s.db.QueryRowContext(context.Background(), ` - SELECT COUNT(*) FROM read_cursors WHERE last_read_message_id = ? - `, messageID).Scan(&count); err != nil { + var referenced bool + if err := s.db.QueryRowContext( + context.Background(), messageHasReadCursorQuery, messageID, messageID, + ).Scan(&referenced); err != nil { return false, fmt.Errorf("check read cursors for message %q: %w", messageID, err) } - return count > 0, nil + return referenced, nil } // MessageHasOpenSend reports whether an unfinished send (an outbox row not yet @@ -79,12 +84,9 @@ func (s *Store) MessageHasReadCursor(messageID string) (bool, error) { // open. func (s *Store) MessageHasOpenSend(messageID string) (bool, error) { var exists bool - if err := s.db.QueryRowContext(context.Background(), ` - SELECT EXISTS ( - SELECT 1 FROM outbox - WHERE local_message_id = ? AND state NOT IN `+closedOutboxStates+` - ) - `, messageID).Scan(&exists); err != nil { + if err := s.db.QueryRowContext( + context.Background(), messageHasOpenSendQuery, messageID, + ).Scan(&exists); err != nil { return false, fmt.Errorf("check open sends for message %q: %w", messageID, err) } return exists, nil @@ -93,6 +95,30 @@ func (s *Store) MessageHasOpenSend(messageID string) (bool, error) { // closedOutboxStates are the terminal outbox states, as an SQL list. const closedOutboxStates = `('confirmed', 'rejected', 'canceled')` +// The open-send lookups seek outbox_local_message_idx on local_message_id. +const ( + messageHasOpenSendQuery = ` + SELECT EXISTS ( + SELECT 1 FROM outbox + WHERE local_message_id = ? AND state NOT IN ` + closedOutboxStates + ` + ) + ` + openSendElsewhereQuery = ` + SELECT outbox_id, state FROM outbox + WHERE local_message_id = ? AND conversation_id <> ? AND state NOT IN ` + closedOutboxStates + ` + ORDER BY outbox_id + LIMIT 1 + ` +) + +const messageHasReadCursorQuery = ` + SELECT EXISTS ( + SELECT 1 FROM read_cursors + WHERE conversation_id = (SELECT conversation_id FROM messages WHERE message_id = ?) + AND last_read_message_id = ? + ) +` + // SelfIdentityID returns one self identity for the account, if any. func (s *Store) SelfIdentityID(accountID string) (string, bool, error) { var id string @@ -426,12 +452,9 @@ func applyRepairDelete( // copy and belongs to a conversation other than conversationID. func refuseOpenSend(ctx context.Context, tx *sql.Tx, messageID, conversationID string) error { var outboxID, state string - err := tx.QueryRowContext(ctx, ` - SELECT outbox_id, state FROM outbox - WHERE local_message_id = ? AND conversation_id <> ? AND state NOT IN `+closedOutboxStates+` - ORDER BY outbox_id - LIMIT 1 - `, messageID, conversationID).Scan(&outboxID, &state) + err := tx.QueryRowContext( + ctx, openSendElsewhereQuery, messageID, conversationID, + ).Scan(&outboxID, &state) switch { case errors.Is(err, sql.ErrNoRows): return nil diff --git a/internal/storage/sqlite/repositories.go b/internal/storage/sqlite/repositories.go index 7dca7970..157068d7 100644 --- a/internal/storage/sqlite/repositories.go +++ b/internal/storage/sqlite/repositories.go @@ -5,7 +5,6 @@ import ( "database/sql" "errors" "fmt" - "sort" "strings" moderncsqlite "modernc.org/sqlite" @@ -753,33 +752,51 @@ func (s *Store) ListConversationsByRecency(accountID string) ([]Conversation, er return conversations, nil } -// ListConversationsByRecencyAllAccounts merges every account's recency list, -// orders the result deterministically, and returns at most limit rows. +// conversationsByRecencyQuery is ListConversationsByRecencyAllAccounts in one +// statement over conversations_recency_idx (archived_at_ms, last_message_at_ms +// DESC, conversation_id). Unarchived conversations are its archived_at_ms IS +// NULL prefix, already in recency order, so that arm stops after limit rows. +// Archived ones are the archived_at_ms >= 0 range (the column's CHECK makes +// that every non-NULL value; IS NOT NULL would plan as a table scan without +// statistics); the range is ordered by archive time first, so that arm reads +// and sorts every archived conversation. The read is therefore bounded by the +// limit plus the archived count, not by the table. +const conversationsByRecencyQuery = ` + SELECT * FROM ( + SELECT ` + conversationColumns + ` + FROM conversations + WHERE archived_at_ms IS NULL + ORDER BY last_message_at_ms DESC, conversation_id + LIMIT ? + ) + UNION ALL + SELECT * FROM ( + SELECT ` + conversationColumns + ` + FROM conversations + WHERE archived_at_ms >= 0 + ORDER BY last_message_at_ms DESC, conversation_id + LIMIT ? + ) + ORDER BY last_message_at_ms DESC, conversation_id + LIMIT ? +` + +// ListConversationsByRecencyAllAccounts returns at most limit conversations +// across every account, newest first, with conversation ID as a deterministic +// tie-breaker. func (s *Store) ListConversationsByRecencyAllAccounts(limit int) ([]Conversation, error) { if limit <= 0 { return []Conversation{}, nil } - accounts, err := s.ListAccounts() + rows, err := s.db.QueryContext( + context.Background(), conversationsByRecencyQuery, limit, limit, limit, + ) if err != nil { return nil, fmt.Errorf("list conversations across accounts: %w", err) } - - conversations := make([]Conversation, 0) - for _, account := range accounts { - accountConversations, err := s.ListConversationsByRecency(account.AccountID) - if err != nil { - return nil, fmt.Errorf("list conversations across accounts: %w", err) - } - conversations = append(conversations, accountConversations...) - } - sort.Slice(conversations, func(i, j int) bool { - if conversations[i].LastMessageAtMS != conversations[j].LastMessageAtMS { - return conversations[i].LastMessageAtMS > conversations[j].LastMessageAtMS - } - return conversations[i].ConversationID < conversations[j].ConversationID - }) - if len(conversations) > limit { - conversations = conversations[:limit] + conversations, err := collectRows(rows, scanConversation) + if err != nil { + return nil, fmt.Errorf("list conversations across accounts: %w", err) } return conversations, nil } diff --git a/internal/v2read/messages.go b/internal/v2read/messages.go index 9a57fa5b..cddc6f35 100644 --- a/internal/v2read/messages.go +++ b/internal/v2read/messages.go @@ -84,35 +84,3 @@ func (s *Source) GetMessagesAroundMessage( } return s.mapMessages(messages) } - -func (s *Source) walkConversationMessages( - conversationID string, - visit func(sqlite.Message), -) error { - var beforeMS int64 - var beforeID string - for { - page, err := s.messages.ListMessagesByConversation( - context.Background(), - conversationID, - beforeMS, - beforeID, - sourceMessagePageSize, - ) - if err != nil { - return err - } - for _, message := range page { - visit(message) - } - if len(page) < sourceMessagePageSize { - return nil - } - last := page[len(page)-1] - if last.OccurredAtMS == beforeMS && last.MessageID == beforeID { - return errors.New("v2 read message pagination did not advance") - } - beforeMS = last.OccurredAtMS - beforeID = last.MessageID - } -} diff --git a/internal/v2read/stats.go b/internal/v2read/stats.go index c9bc2d4d..e1fb6ea3 100644 --- a/internal/v2read/stats.go +++ b/internal/v2read/stats.go @@ -9,6 +9,12 @@ import ( "github.com/maxghenis/openmessage/internal/storage/sqlite" ) +// Every aggregate below is per account: a COUNT(*) range or MAX seeks on an +// account-leading index (see sqlite.Store.LatestMessageTimes). Before, each +// loaded the account's whole conversation list; PlatformStats and MessageCount +// then paged through every message, and LatestTimestamp read each +// conversation's newest message. + // MessageCount returns all message rows or those whose account maps to the // requested legacy source platform. func (s *Source) MessageCount(sourcePlatform string) (int, error) { @@ -24,17 +30,11 @@ func (s *Source) MessageCount(sourcePlatform string) (int, error) { if sourcePlatform != "" && platformForBridgeKey(account.BridgeKey) != sourcePlatform { continue } - conversations, err := s.store.ListConversationsByRecency(account.AccountID) + accountCount, err := s.store.CountMessages(account.AccountID) if err != nil { return 0, err } - for _, conversation := range conversations { - if err := s.walkConversationMessages(conversation.ConversationID, func(sqlite.Message) { - count++ - }); err != nil { - return 0, err - } - } + count += int(accountCount) } return count, nil } @@ -54,11 +54,11 @@ func (s *Source) ConversationCount(sourcePlatform string) (int, error) { if sourcePlatform != "" && platformForBridgeKey(account.BridgeKey) != sourcePlatform { continue } - conversations, err := s.store.ListConversationsByRecency(account.AccountID) + accountCount, err := s.store.CountConversations(account.AccountID) if err != nil { return 0, err } - count += len(conversations) + count += int(accountCount) } return count, nil } @@ -77,27 +77,40 @@ func (s *Source) LatestTimestamp(sourcePlatform string) (int64, error) { if platformForBridgeKey(account.BridgeKey) != sourcePlatform { continue } - conversations, err := s.store.ListConversationsByRecency(account.AccountID) + accountLatest, _, err := s.store.LatestMessageTimes(account.AccountID) if err != nil { return 0, err } - for _, conversation := range conversations { - messages, err := s.messages.ListMessagesByConversation( - context.Background(), conversation.ConversationID, 0, "", 1, - ) - if err != nil { - return 0, err - } - if len(messages) > 0 && messages[0].OccurredAtMS > latest { - latest = messages[0].OccurredAtMS - } - } + latest = max(latest, accountLatest) } return latest, nil } -// PlatformStats summarizes v2 messages under their legacy platform labels. +// PlatformStats summarizes v2 messages under their legacy platform labels. A +// platform appears only when it holds at least one message. func (s *Source) PlatformStats() ([]db.PlatformStat, error) { + return s.platformStats(true) +} + +// PlatformLatest is PlatformStats without the message counts: only seeks, so +// it stays cheap however large the store grows. /api/status freshness uses it. +func (s *Source) PlatformLatest() ([]db.PlatformLatest, error) { + stats, err := s.platformStats(false) + if err != nil { + return nil, err + } + latest := make([]db.PlatformLatest, 0, len(stats)) + for _, stat := range stats { + latest = append(latest, db.PlatformLatest{ + Platform: stat.Platform, + LatestMS: stat.LatestMS, + LatestRecvMS: stat.LatestRecvMS, + }) + } + return latest, nil +} + +func (s *Source) platformStats(withCounts bool) ([]db.PlatformStat, error) { if err := s.ready(); err != nil { return nil, err } @@ -107,33 +120,33 @@ func (s *Source) PlatformStats() ([]db.PlatformStat, error) { } byPlatform := make(map[string]*db.PlatformStat) for _, account := range accounts { - platform := platformForBridgeKey(account.BridgeKey) - if strings.TrimSpace(platform) == "" { - platform = "unknown" - } - conversations, err := s.store.ListConversationsByRecency(account.AccountID) + latestMS, latestRecvMS, err := s.store.LatestMessageTimes(account.AccountID) if err != nil { return nil, err } - for _, conversation := range conversations { - if err := s.walkConversationMessages(conversation.ConversationID, func(message sqlite.Message) { - stat := byPlatform[platform] - if stat == nil { - stat = &db.PlatformStat{Platform: platform} - byPlatform[platform] = stat - } - stat.Count++ - if message.OccurredAtMS > stat.LatestMS { - stat.LatestMS = message.OccurredAtMS - } - if message.Direction == sqlite.MessageDirectionIncoming && - message.OccurredAtMS > stat.LatestRecvMS { - stat.LatestRecvMS = message.OccurredAtMS - } - }); err != nil { + // occurred_at_ms is always positive, so 0 means the account has no + // messages; it then contributes no platform row. + if latestMS == 0 { + continue + } + var count int64 + if withCounts { + if count, err = s.store.CountMessages(account.AccountID); err != nil { return nil, err } } + platform := platformForBridgeKey(account.BridgeKey) + if strings.TrimSpace(platform) == "" { + platform = "unknown" + } + stat := byPlatform[platform] + if stat == nil { + stat = &db.PlatformStat{Platform: platform} + byPlatform[platform] = stat + } + stat.Count += int(count) + stat.LatestMS = max(stat.LatestMS, latestMS) + stat.LatestRecvMS = max(stat.LatestRecvMS, latestRecvMS) } stats := make([]db.PlatformStat, 0, len(byPlatform)) for _, stat := range byPlatform { diff --git a/internal/v2read/stats_property_test.go b/internal/v2read/stats_property_test.go new file mode 100644 index 00000000..335fdf00 --- /dev/null +++ b/internal/v2read/stats_property_test.go @@ -0,0 +1,187 @@ +package v2read + +import ( + "fmt" + "math/rand" + "reflect" + "sort" + "strings" + "testing" + "testing/quick" + + "github.com/maxghenis/openmessage/internal/db" + "github.com/maxghenis/openmessage/internal/storage/sqlite" +) + +type statsCase struct { + Seed int64 +} + +func (statsCase) Generate(r *rand.Rand, _ int) reflect.Value { + return reflect.ValueOf(statsCase{Seed: r.Int63()}) +} + +// Invariants of the per-account aggregates behind the status surfaces, for +// random stores where several accounts share a platform and one bridge key +// has no mapping of its own: +// - a platform appears in PlatformStats exactly when it holds a message; +// - Count is its number of messages, LatestMS the newest occurred_at_ms and +// LatestRecvMS the newest incoming one (0 when it received nothing); +// - PlatformLatest is PlatformStats without Count, in the same order; +// - MessageCount, ConversationCount and LatestTimestamp agree with the same +// sums and maxima, per platform and in total. +func TestStatsMatchReferenceAggregatesProperty(t *testing.T) { + // "\t" passes the accounts.bridge_key CHECK (SQLite's trim strips only + // spaces) but strings.TrimSpace blanks it, so it is the "unknown" platform. + bridges := []string{"google_messages", "google_messages", "signal_cli", "whatsmeow", "mystery_bridge", "\t"} + property := func(c statsCase) bool { + r := rand.New(rand.NewSource(c.Seed)) + store, messages, source := openSourceTestStore(t) + type account struct{ id, platform string } + var accounts []account + for i, bridge := range bridges { + if r.Intn(3) == 0 { + continue + } + id := fmt.Sprintf("account-%d", i) + seedSourceAccount(t, store, id, bridge) + // An unmapped bridge key is its own platform label; a blank one is + // "unknown" in PlatformStats. + platform := platformForBridgeKey(bridge) + if strings.TrimSpace(platform) == "" { + platform = "unknown" + } + accounts = append(accounts, account{id: id, platform: platform}) + } + if len(accounts) == 0 { + return true + } + conversationsPer := map[string]int{} + var conversations []sqlite.Conversation + for i := 0; i < 1+r.Intn(6); i++ { + owner := accounts[r.Intn(len(accounts))] + conversation := sqlite.Conversation{ + ConversationID: fmt.Sprintf("conversation-%d", i), + AccountID: owner.id, + RemoteConversationID: fmt.Sprintf("remote-%d", i), + Kind: sqlite.ConversationKindDirect, + NotificationMode: sqlite.NotificationModeAll, + } + seedSourceConversation(t, store, conversation) + conversations = append(conversations, conversation) + conversationsPer[owner.platform]++ + } + want := map[string]*db.PlatformStat{} + for i := 0; i < r.Intn(40); i++ { + conversation := conversations[r.Intn(len(conversations))] + direction := sqlite.MessageDirectionIncoming + if r.Intn(2) == 0 { + direction = sqlite.MessageDirectionOutgoing + } + occurred := 1 + r.Int63n(50) + importSourceMessage(t, messages, sqlite.Message{ + MessageID: fmt.Sprintf("message-%d", i), + ConversationID: conversation.ConversationID, + AccountID: conversation.AccountID, + RemoteMessageID: fmt.Sprintf("remote-message-%d", i), + Direction: direction, + // Edited and deleted rows still count: the aggregates cover + // every stored message, as the walk they replace did. + State: []sqlite.MessageState{sqlite.MessageStateActive, sqlite.MessageStateEdited, sqlite.MessageStateDeleted}[r.Intn(3)], + OccurredAtMS: occurred, + }) + var platform string + for _, a := range accounts { + if a.id == conversation.AccountID { + platform = a.platform + } + } + stat := want[platform] + if stat == nil { + stat = &db.PlatformStat{Platform: platform} + want[platform] = stat + } + stat.Count++ + stat.LatestMS = max(stat.LatestMS, occurred) + if direction == sqlite.MessageDirectionIncoming { + stat.LatestRecvMS = max(stat.LatestRecvMS, occurred) + } + } + wantStats := []db.PlatformStat{} + for _, stat := range want { + wantStats = append(wantStats, *stat) + } + sort.Slice(wantStats, func(i, j int) bool { + if wantStats[i].LatestMS != wantStats[j].LatestMS { + return wantStats[i].LatestMS > wantStats[j].LatestMS + } + return wantStats[i].Platform < wantStats[j].Platform + }) + + stats, err := source.PlatformStats() + if err != nil || !reflect.DeepEqual(stats, wantStats) { + t.Errorf("seed %d: PlatformStats() = %+v, %v; want %+v", c.Seed, stats, err, wantStats) + return false + } + latest, err := source.PlatformLatest() + if err != nil || len(latest) != len(wantStats) { + t.Errorf("seed %d: PlatformLatest() = %+v, %v; want %d rows", c.Seed, latest, err, len(wantStats)) + return false + } + for i, stat := range wantStats { + if latest[i] != (db.PlatformLatest{Platform: stat.Platform, LatestMS: stat.LatestMS, LatestRecvMS: stat.LatestRecvMS}) { + t.Errorf("seed %d: PlatformLatest()[%d] = %+v, want %+v", c.Seed, i, latest[i], stat) + return false + } + } + platforms := []string{"sms", "signal", "whatsapp", "imessage"} + for platform := range want { + platforms = append(platforms, platform) + } + total := 0 + for _, stat := range want { + total += stat.Count + } + for _, platform := range platforms { + var wantCount int + var wantLatest int64 + wantConversations := conversationsPer[platform] + if stat := want[platform]; stat != nil { + wantCount, wantLatest = stat.Count, stat.LatestMS + } + if platform == "unknown" { + // The per-platform counters filter on the raw label, which is + // empty for a blank bridge key; only PlatformStats relabels it + // "unknown". Unchanged by the per-account rewrite. + wantCount, wantLatest, wantConversations = 0, 0, 0 + } + count, err := source.MessageCount(platform) + if err != nil || count != wantCount { + t.Errorf("seed %d: MessageCount(%q) = %d, %v; want %d", c.Seed, platform, count, err, wantCount) + return false + } + latestMS, err := source.LatestTimestamp(platform) + if err != nil || latestMS != wantLatest { + t.Errorf("seed %d: LatestTimestamp(%q) = %d, %v; want %d", c.Seed, platform, latestMS, err, wantLatest) + return false + } + conversations, err := source.ConversationCount(platform) + if err != nil || conversations != wantConversations { + t.Errorf("seed %d: ConversationCount(%q) = %d, %v; want %d", c.Seed, platform, conversations, err, wantConversations) + return false + } + } + if count, err := source.MessageCount(""); err != nil || count != total { + t.Errorf("seed %d: MessageCount(\"\") = %d, %v; want %d", c.Seed, count, err, total) + return false + } + if count, err := source.ConversationCount(""); err != nil || count != len(conversations) { + t.Errorf("seed %d: ConversationCount(\"\") = %d, %v; want %d", c.Seed, count, err, len(conversations)) + return false + } + return true + } + if err := quick.Check(property, &quick.Config{MaxCount: 40}); err != nil { + t.Fatal(err) + } +} diff --git a/internal/web/api.go b/internal/web/api.go index 78a6781e..7a916914 100644 --- a/internal/web/api.go +++ b/internal/web/api.go @@ -238,9 +238,10 @@ func APIHandlerWithOptions(store *db.Store, cli *client.Client, logger zerolog.L } // Per-platform data-freshness, used to catch "zombie" bridges that report // connected=true while no longer actually syncing (the connection flag - // lies; the data doesn't). Computing this scans the messages table, and - // /api/status is polled every few seconds, so cache it — staleness is a - // multi-day signal, so a 30s cache is plenty fresh. + // lies; the data doesn't). /api/status is polled every few seconds, so + // cache it — staleness is a multi-day signal, so a 30s cache is plenty + // fresh. On v2 the read is a few index seeks (platformLatest); the legacy + // store's PlatformStats still aggregates its whole messages table. var ( freshnessMu sync.Mutex freshnessComputed time.Time @@ -253,7 +254,7 @@ func APIHandlerWithOptions(store *db.Store, cli *client.Client, logger zerolog.L if freshnessValue != nil && time.Since(freshnessComputed) < 30*time.Second { return freshnessValue } - stats, err := reads.PlatformStats() + stats, err := platformLatest(reads) if err != nil { return freshnessValue // keep last good value on error } @@ -3815,6 +3816,34 @@ func daysBehind(older, newer int64) int { // disagreement are normal; a real projection stall shows up as minutes. const projectionLagFloorMS = int64(5 * 60 * 1000) +// platformLatestSource is implemented by read sources that report each +// platform's newest message times without counting every message (v2read). +type platformLatestSource interface { + PlatformLatest() ([]db.PlatformLatest, error) +} + +// platformLatest returns each stored platform's newest sent/received times. +// Freshness never reads the counts, so a v2 store answers with index seeks; the +// legacy store has no such method and still runs its GROUP BY. +func platformLatest(reads readsource.ReadSource) ([]db.PlatformLatest, error) { + if source, ok := reads.(platformLatestSource); ok { + return source.PlatformLatest() + } + stats, err := reads.PlatformStats() + if err != nil { + return nil, err + } + latest := make([]db.PlatformLatest, 0, len(stats)) + for _, st := range stats { + latest = append(latest, db.PlatformLatest{ + Platform: st.Platform, + LatestMS: st.LatestMS, + LatestRecvMS: st.LatestRecvMS, + }) + } + return latest, nil +} + // addProjectionLag annotates each per-platform freshness entry with how far the // active read source trails the legacy store, and stamps a top-level // `projection_stalled` flag when any platform is behind. Only meaningful in diff --git a/internal/web/freshness_latest_test.go b/internal/web/freshness_latest_test.go new file mode 100644 index 00000000..a097e6d9 --- /dev/null +++ b/internal/web/freshness_latest_test.go @@ -0,0 +1,77 @@ +package web + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/rs/zerolog" + + "github.com/maxghenis/openmessage/internal/db" + "github.com/maxghenis/openmessage/internal/v2read" +) + +// The v2 read source must keep the count-free freshness read; without it every +// /api/status freshness refresh falls back to PlatformStats and counts every +// message. +var _ platformLatestSource = (*v2read.Source)(nil) + +// latestOnlyReads answers freshness from PlatformLatest and fails the test if +// anything asks it to count messages. +type latestOnlyReads struct { + stubReads + t *testing.T + latest []db.PlatformLatest +} + +func (s latestOnlyReads) PlatformLatest() ([]db.PlatformLatest, error) { return s.latest, nil } + +func (s latestOnlyReads) PlatformStats() ([]db.PlatformStat, error) { + s.t.Errorf("/api/status freshness called PlatformStats; it must use PlatformLatest") + return nil, nil +} + +func TestStatusFreshnessReadsPlatformLatestWithoutCounting(t *testing.T) { + store, err := db.New(":memory:") + if err != nil { + t.Fatal(err) + } + defer store.Close() + + now := time.Now().UnixMilli() + fourDaysAgo := now - 4*24*60*60*1000 + reads := latestOnlyReads{t: t, latest: []db.PlatformLatest{ + {Platform: "whatsapp", LatestMS: now, LatestRecvMS: now - 1}, + {Platform: "signal", LatestMS: fourDaysAgo, LatestRecvMS: fourDaysAgo - 1}, + }} + h := APIHandlerWithOptions(store, nil, zerolog.Nop(), nil, APIOptions{Reads: reads}) + srv := httptest.NewServer(h) + defer srv.Close() + + resp, err := http.Get(srv.URL + "/api/status") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + var payload map[string]any + if err := json.NewDecoder(resp.Body).Decode(&payload); err != nil { + t.Fatal(err) + } + freshness, ok := payload["freshness"].(map[string]any) + if !ok { + t.Fatalf("freshness block = %v", payload["freshness"]) + } + if newest, _ := freshness["newest_ms"].(float64); int64(newest) != now { + t.Fatalf("newest_ms = %v, want %d", freshness["newest_ms"], now) + } + whatsapp, _ := freshness["whatsapp"].(map[string]any) + if received, _ := whatsapp["latest_received_ms"].(float64); int64(received) != now-1 { + t.Fatalf("whatsapp freshness = %v, want latest_received_ms %d", whatsapp, now-1) + } + signal, _ := freshness["signal"].(map[string]any) + if stale, _ := signal["stale"].(bool); !stale { + t.Fatalf("signal freshness = %v, want stale (four days behind whatsapp)", signal) + } +}