diff --git a/internal/store/account_identities.go b/internal/store/account_identities.go index 91eefcd93..3789a087b 100644 --- a/internal/store/account_identities.go +++ b/internal/store/account_identities.go @@ -270,6 +270,18 @@ func (s *Store) mergeAccountIdentitySignalsTx( // the same writer-locked transaction as the write, so a removal that lands // after the caller read the confirmed set cannot be undone. It reports whether // a row was inserted and whether one was present to merge into. +// +// The lookup keys on address_key, the persisted comparison-canonical form, +// so both backends match under the same Go-owned rule and the partial unique +// index on (source_id, address_key) can reject a concurrent case-variant +// insert (the retry loop in the callers then re-reads and merges). Rows +// written by binaries that predate the column carry address_key = ” until +// the next store open repairs them; the fallback predicate matches those +// under the legacy case-aware rule so an in-session legacy row is merged +// into (and promoted to keyed) rather than shadowed by an insert that would +// then collide on the raw-bytes primary key. ORDER BY prefers the keyed row +// when a legacy duplicate coexists; the unique index guarantees at most one +// keyed row per key. func (s *Store) mergeAccountIdentitySignalsTxWith( ctx context.Context, tx *loggedTx, @@ -279,11 +291,14 @@ func (s *Store) mergeAccountIdentitySignalsTxWith( match identifierMatch, allowInsert bool, ) (inserted, present bool, err error) { - whereAddr := match.WhereClause("address") - var existing string - selectSQL := `SELECT source_signal FROM account_identities - WHERE source_id = ? AND ` + whereAddr + s.dialect.SelectForUpdate() - err = tx.QueryRowContext(ctx, selectSQL, sourceID, match.BindValue()).Scan(&existing) + key := NormalizeIdentifierForCompare(addr) + var existingAddr, existing string + selectSQL := `SELECT address, source_signal FROM account_identities + WHERE source_id = ? AND (address_key = ? + OR (address_key = '' AND ` + match.WhereClause("address") + `)) + ORDER BY address_key DESC LIMIT 1` + s.dialect.SelectForUpdate() + err = tx.QueryRowContext(ctx, selectSQL, sourceID, key, match.BindValue()). + Scan(&existingAddr, &existing) switch { case errors.Is(err, sql.ErrNoRows): if !allowInsert { @@ -294,9 +309,9 @@ func (s *Store) mergeAccountIdentitySignalsTxWith( merged = mergeSignalSet(merged, signal) } if _, err := tx.ExecContext(ctx, - `INSERT INTO account_identities (source_id, address, source_signal) - VALUES (?, ?, ?)`, - sourceID, addr, merged, + `INSERT INTO account_identities (source_id, address, address_key, source_signal) + VALUES (?, ?, ?, ?)`, + sourceID, addr, key, merged, ); err != nil { return false, false, fmt.Errorf("insert account identity: %w", err) } @@ -310,9 +325,9 @@ func (s *Store) mergeAccountIdentitySignalsTxWith( } if merged != existing { if _, err := tx.ExecContext(ctx, - `UPDATE account_identities SET source_signal = ? - WHERE source_id = ? AND `+whereAddr, - merged, sourceID, match.BindValue(), + `UPDATE account_identities SET source_signal = ?, address_key = ? + WHERE source_id = ? AND address = ?`, + merged, NormalizeIdentifierForCompare(existingAddr), sourceID, existingAddr, ); err != nil { return false, true, fmt.Errorf("update source_signal: %w", err) } diff --git a/internal/store/dialect_pg.go b/internal/store/dialect_pg.go index 2c0792a2b..4b0408be8 100644 --- a/internal/store/dialect_pg.go +++ b/internal/store/dialect_pg.go @@ -652,6 +652,7 @@ func (d *PostgreSQLDialect) LegacyColumnMigrations() []ColumnMigration { {`ALTER TABLE conversations ADD COLUMN IF NOT EXISTS title TEXT`, "title"}, {`ALTER TABLE conversations ADD COLUMN IF NOT EXISTS conversation_type TEXT NOT NULL DEFAULT 'email_thread'`, "conversation_type"}, {`ALTER TABLE labels ADD COLUMN IF NOT EXISTS system_role TEXT`, "labels.system_role"}, + {`ALTER TABLE account_identities ADD COLUMN IF NOT EXISTS address_key TEXT NOT NULL DEFAULT ''`, "account_identities.address_key"}, {`ALTER TABLE participant_identifiers ADD COLUMN IF NOT EXISTS service_id BIGINT REFERENCES communication_services(id) ON DELETE SET NULL`, "pi_service_id"}, {`ALTER TABLE participant_identifiers ADD COLUMN IF NOT EXISTS scope_kind TEXT`, "pi_scope_kind"}, {`ALTER TABLE participant_identifiers ADD COLUMN IF NOT EXISTS scope_value TEXT`, "pi_scope_value"}, diff --git a/internal/store/dialect_sqlite.go b/internal/store/dialect_sqlite.go index 0fb2f7a4c..79f718ccd 100644 --- a/internal/store/dialect_sqlite.go +++ b/internal/store/dialect_sqlite.go @@ -1953,6 +1953,7 @@ func (d *SQLiteDialect) LegacyColumnMigrations() []ColumnMigration { {`ALTER TABLE conversations ADD COLUMN title TEXT`, "title"}, {`ALTER TABLE conversations ADD COLUMN conversation_type TEXT NOT NULL DEFAULT 'email_thread'`, "conversation_type"}, {`ALTER TABLE labels ADD COLUMN system_role TEXT`, "labels.system_role"}, + {`ALTER TABLE account_identities ADD COLUMN address_key TEXT NOT NULL DEFAULT ''`, "account_identities.address_key"}, {`ALTER TABLE participant_identifiers ADD COLUMN service_id INTEGER REFERENCES communication_services(id) ON DELETE SET NULL`, "pi_service_id"}, {`ALTER TABLE participant_identifiers ADD COLUMN scope_kind TEXT`, "pi_scope_kind"}, {`ALTER TABLE participant_identifiers ADD COLUMN scope_value TEXT`, "pi_scope_value"}, diff --git a/internal/store/migrate_account_identity_key.go b/internal/store/migrate_account_identity_key.go new file mode 100644 index 000000000..8b40d879d --- /dev/null +++ b/internal/store/migrate_account_identity_key.go @@ -0,0 +1,247 @@ +package store + +import ( + "context" + "fmt" + "sort" + "strings" + "time" +) + +// ensureAccountIdentityAddressKeys brings account_identities in line with the +// persisted-comparison-key invariant: every row's address_key holds +// NormalizeIdentifierForCompare(address), case-variant duplicates of one +// logical identity are merged into a single row, and a partial unique index +// on (source_id, address_key) enforces the invariant at the schema level for +// every keyed writer, whichever process or backend it comes from. +// +// It runs on every store open, after LegacyColumnMigrations has added the +// column. The gate is the scan itself, not a run-once ledger entry: a +// previous-release binary writes rows through the column's ” default, so +// "done once" can go stale at any time, while the scan is cheap because the +// table holds only the confirmed "me" identities per source. Rows already +// carrying their derived key cost one read and no write. +// +// The repair takes the identity-mutation row lock first, mirroring +// AddAccountIdentity / RemoveAccountIdentity / MigrateLegacyIdentityConfig, +// so it serializes with every supported identity writer across processes. +// When duplicate rows collapse, the survivor keeps the earliest confirmed_at +// and the union of the signal sets, message attribution for the affected +// sources is recomputed in the same transaction, and both identity revisions +// are bumped (a collapse changes the exported identity row set and can +// change owner-participant derivation). Key-only backfills touch no derived +// state: comparison consumers still read address, so nothing they produce +// changes. +func (s *Store) ensureAccountIdentityAddressKeys(ctx context.Context) error { + needsRepair, err := s.accountIdentityKeysNeedRepair(ctx) + if err != nil { + return err + } + if needsRepair { + if err := s.repairAccountIdentityAddressKeys(ctx); err != nil { + return err + } + } + // The index is created after the repair because CREATE UNIQUE INDEX + // fails while case-variant duplicates still share a derived key. It + // lives here rather than in schema.sql/schema_pg.sql because on an + // upgraded archive those scripts execute before the legacy-column loop + // has added address_key. The WHERE clause exempts the '' sentinel: + // previous-release inserts land with the column default and must not + // collide with each other; the next open keys them through the repair + // above. Ledgered so the DDL runs once per archive, and built through + // the maintenance escape hatch so a lock held by a concurrent identity + // writer cannot trip the pool-wide PostgreSQL statement timeout and + // fail the open; IF NOT EXISTS covers a cancellation between the + // create and the ledger write. + return s.runOnceMigration(ctx, migrationAccountIdentityAddressKeyIndex, false, + func(ctx context.Context) error { + return s.runMaintenance(ctx, func(ctx context.Context, tx *loggedTx) error { + if _, err := tx.ExecContext(ctx, ` + CREATE UNIQUE INDEX IF NOT EXISTS idx_account_identities_address_key + ON account_identities(source_id, address_key) + WHERE address_key <> '' + `); err != nil { + return fmt.Errorf("create account identity address key index: %w", err) + } + return nil + }) + }) +} + +// accountIdentityKeysNeedRepair reports whether any row's stored key differs +// from its derived key. Read-only fast path for every open; the shape check +// runs in Go because looksLikeEmail is Go-owned and deliberately has no SQL +// reimplementation. +func (s *Store) accountIdentityKeysNeedRepair(ctx context.Context) (bool, error) { + rows, err := s.db.QueryContext(ctx, + `SELECT address, address_key FROM account_identities`) + if err != nil { + return false, fmt.Errorf("scan account identity keys: %w", err) + } + defer func() { _ = rows.Close() }() + for rows.Next() { + var address, key string + if err := rows.Scan(&address, &key); err != nil { + return false, fmt.Errorf("scan account identity key row: %w", err) + } + if key != NormalizeIdentifierForCompare(address) { + return true, nil + } + } + return false, rows.Err() +} + +// accountIdentityKeyRow is one account_identities row as read by the repair. +type accountIdentityKeyRow struct { + sourceID int64 + address string + addressKey string + signals string + confirmedAt time.Time +} + +// accountIdentityGroupKey identifies one logical identity: a source and the +// derived comparison key its rows share. +type accountIdentityGroupKey struct { + sourceID int64 + key string +} + +// repairAccountIdentityAddressKeys runs through the maintenance escape hatch: +// a duplicate collapse refreshes source-wide message attribution, whose cost +// scales with archive size, and the ordinary pool-wide PostgreSQL statement +// timeout would cancel it on a large upgraded source and fail every +// subsequent open. The identity-mutation lock is taken first, matching the +// lock order of every other identity writer. +func (s *Store) repairAccountIdentityAddressKeys(ctx context.Context) error { + return s.runMaintenance(ctx, func(ctx context.Context, tx *loggedTx) error { + if err := s.lockIdentityMutationTxContext(ctx, tx); err != nil { + return err + } + // Re-read under the lock: the lock-free fast path may have raced a + // writer, and the repair must act on the committed state it now owns. + all, err := readAccountIdentityKeyRows(ctx, tx) + if err != nil { + return err + } + + groups := make(map[accountIdentityGroupKey][]accountIdentityKeyRow) + order := make([]accountIdentityGroupKey, 0, len(all)) + for _, row := range all { + gk := accountIdentityGroupKey{ + sourceID: row.sourceID, + key: NormalizeIdentifierForCompare(row.address), + } + if _, seen := groups[gk]; !seen { + order = append(order, gk) + } + groups[gk] = append(groups[gk], row) + } + + collapsed := false + collapsedSources := make(map[int64]struct{}) + for _, gk := range order { + group := groups[gk] + if len(group) == 1 && group[0].addressKey == gk.key { + continue + } + survivor := pickAccountIdentitySurvivor(group, gk.key) + mergedSignals := "" + earliest := survivor.confirmedAt + for _, row := range group { + for signal := range strings.SplitSeq(row.signals, ",") { + if signal != "" { + mergedSignals = mergeSignalSet(mergedSignals, signal) + } + } + if row.confirmedAt.Before(earliest) { + earliest = row.confirmedAt + } + } + for _, row := range group { + if row.address == survivor.address { + continue + } + if _, err := tx.ExecContext(ctx, + `DELETE FROM account_identities WHERE source_id = ? AND address = ?`, + row.sourceID, row.address, + ); err != nil { + return fmt.Errorf("collapse duplicate account identity: %w", err) + } + collapsed = true + collapsedSources[row.sourceID] = struct{}{} + } + if _, err := tx.ExecContext(ctx, + `UPDATE account_identities + SET address_key = ?, source_signal = ?, confirmed_at = ? + WHERE source_id = ? AND address = ?`, + gk.key, mergedSignals, earliest, survivor.sourceID, survivor.address, + ); err != nil { + return fmt.Errorf("backfill account identity key: %w", err) + } + } + + if !collapsed { + return nil + } + if _, err := s.bumpIdentityRevisionContext(ctx, tx); err != nil { + return err + } + if err := s.bumpAccountIdentityRevisionContext(ctx, tx); err != nil { + return err + } + for sourceID := range collapsedSources { + if err := refreshSourceMessageAttributionContext(ctx, tx, sourceID, ""); err != nil { + return fmt.Errorf("refresh attribution after identity collapse (source=%d): %w", sourceID, err) + } + } + return nil + }) +} + +func readAccountIdentityKeyRows( + ctx context.Context, tx *loggedTx, +) ([]accountIdentityKeyRow, error) { + rows, err := tx.QueryContext(ctx, ` + SELECT source_id, address, address_key, source_signal, confirmed_at + FROM account_identities`) + if err != nil { + return nil, fmt.Errorf("read account identity rows: %w", err) + } + defer func() { _ = rows.Close() }() + var out []accountIdentityKeyRow + for rows.Next() { + var row accountIdentityKeyRow + if err := rows.Scan( + &row.sourceID, &row.address, &row.addressKey, + &row.signals, &row.confirmedAt, + ); err != nil { + return nil, fmt.Errorf("scan account identity row: %w", err) + } + out = append(out, row) + } + return out, rows.Err() +} + +// pickAccountIdentitySurvivor chooses which duplicate row keeps the logical +// identity. A row already carrying the correct key wins (it is the one the +// unique index vouches for and the one keyed lookups have been matching); +// otherwise the earliest-confirmed row wins, with the lexicographically +// smallest address as a deterministic tie-break. +func pickAccountIdentitySurvivor( + group []accountIdentityKeyRow, want string, +) accountIdentityKeyRow { + sorted := make([]accountIdentityKeyRow, len(group)) + copy(sorted, group) + sort.Slice(sorted, func(i, j int) bool { + if (sorted[i].addressKey == want) != (sorted[j].addressKey == want) { + return sorted[i].addressKey == want + } + if !sorted[i].confirmedAt.Equal(sorted[j].confirmedAt) { + return sorted[i].confirmedAt.Before(sorted[j].confirmedAt) + } + return sorted[i].address < sorted[j].address + }) + return sorted[0] +} diff --git a/internal/store/migrate_account_identity_key_internal_test.go b/internal/store/migrate_account_identity_key_internal_test.go new file mode 100644 index 000000000..40e7f2656 --- /dev/null +++ b/internal/store/migrate_account_identity_key_internal_test.go @@ -0,0 +1,63 @@ +package store + +import ( + "path/filepath" + "strings" + "sync" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestConcurrentTwoStoreCaseVariantAdds covers issue #311's motivating +// scenario: two Store handles on the same database (the serve-plus-CLI +// deployment) concurrently confirm case variants of one email identity. +// The keyed lookup plus the partial unique index must leave exactly one +// row carrying the union of both signal sets. +func TestConcurrentTwoStoreCaseVariantAdds(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + dbPath := filepath.Join(t.TempDir(), "race.db") + + first, err := OpenForTest(dbPath) + require.NoError(err, "open first store") + t.Cleanup(func() { _ = first.Close() }) + require.NoError(first.InitSchema(), "init schema") + + second, err := OpenForTest(dbPath) + require.NoError(err, "open second store") + t.Cleanup(func() { _ = second.Close() }) + + src, err := first.GetOrCreateSource("gmail", "two-store-race@example.com") + require.NoError(err, "GetOrCreateSource") + + stores := []*Store{first, second} + variants := []string{"Alice@Example.com", "alice@example.com"} + signals := []string{"manual", "account-identifier", "header"} + + const n = 12 + var wg sync.WaitGroup + errs := make([]error, n) + for i := range n { + wg.Add(1) + go func(idx int) { + defer wg.Done() + errs[idx] = stores[idx%2].AddAccountIdentity( + src.ID, variants[idx%2], signals[idx%len(signals)], + ) + }(i) + } + wg.Wait() + for i, err := range errs { + require.NoError(err, "add %d", i) + } + + identities, err := first.ListAccountIdentities(src.ID) + require.NoError(err, "ListAccountIdentities") + require.Len(identities, 1, "case variants across two stores must land in one row") + got := strings.Split(identities[0].SourceSignal, ",") + for _, want := range signals { + assert.Contains(got, want, "merged signal set missing %q", want) + } +} diff --git a/internal/store/migrate_account_identity_key_test.go b/internal/store/migrate_account_identity_key_test.go new file mode 100644 index 000000000..deb266f37 --- /dev/null +++ b/internal/store/migrate_account_identity_key_test.go @@ -0,0 +1,215 @@ +package store_test + +import ( + "database/sql" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "go.kenn.io/msgvault/internal/store" + "go.kenn.io/msgvault/internal/testutil" + "go.kenn.io/msgvault/internal/testutil/storetest" +) + +func TestAccountIdentityAddressKeyBackfillMergesCaseVariantEmails(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + st := testutil.NewTestStore(t) + src, err := st.GetOrCreateSource("gmail", "backfill-case@example.com") + require.NoError(err, "GetOrCreateSource") + + // Two case variants of one logical email identity, written as a + // previous-release binary would: address_key omitted, so it lands as ''. + _, err = st.DB().Exec(st.Rebind( + `INSERT INTO account_identities (source_id, address, source_signal, confirmed_at) + VALUES (?, ?, ?, ?)`), + src.ID, "Alice@Example.com", "manual", "2024-01-02 03:04:05+00:00") + require.NoError(err, "seed first case variant") + _, err = st.DB().Exec(st.Rebind( + `INSERT INTO account_identities (source_id, address, source_signal, confirmed_at) + VALUES (?, ?, ?, ?)`), + src.ID, "alice@example.com", "header", "2024-05-06 07:08:09+00:00") + require.NoError(err, "seed second case variant") + + require.NoError(st.InitSchema(), "reinit schema to run the key repair") + + identities, err := st.ListAccountIdentities(src.ID) + require.NoError(err, "ListAccountIdentities") + require.Len(identities, 1, "case variants must collapse to one row") + got := identities[0] + assert.Equal("Alice@Example.com", got.Address, + "the earliest-confirmed row keeps the logical identity and its display casing") + assert.Equal("header,manual", got.SourceSignal, "signal sets must union") + + var key string + require.NoError(st.DB().QueryRow(st.Rebind( + `SELECT address_key FROM account_identities WHERE source_id = ? AND address = ?`), + src.ID, "Alice@Example.com").Scan(&key)) + assert.Equal("alice@example.com", key, "backfilled key must be the comparison-canonical form") +} + +func TestAccountIdentityAddressKeyBackfillPreservesNonEmailCase(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + st := testutil.NewTestStore(t) + src, err := st.GetOrCreateSource("matrix", "backfill-mxid@example.com") + require.NoError(err, "GetOrCreateSource") + + _, err = st.DB().Exec(st.Rebind( + `INSERT INTO account_identities (source_id, address, source_signal) + VALUES (?, ?, ?)`), + src.ID, "@User:server.org", "manual") + require.NoError(err, "seed mixed-case MXID") + _, err = st.DB().Exec(st.Rebind( + `INSERT INTO account_identities (source_id, address, source_signal) + VALUES (?, ?, ?)`), + src.ID, "@user:server.org", "manual") + require.NoError(err, "seed lowercase MXID") + + require.NoError(st.InitSchema(), "reinit schema to run the key repair") + + identities, err := st.ListAccountIdentities(src.ID) + require.NoError(err, "ListAccountIdentities") + require.Len(identities, 2, "non-email identifiers are case-sensitive and must not merge") + for _, ai := range identities { + var key string + require.NoError(st.DB().QueryRow(st.Rebind( + `SELECT address_key FROM account_identities WHERE source_id = ? AND address = ?`), + src.ID, ai.Address).Scan(&key)) + assert.Equal(ai.Address, key, "non-email key must preserve case verbatim") + } +} + +func TestAccountIdentityKeyIndexRejectsKeyedCaseVariantDuplicate(t *testing.T) { + require := require.New(t) + st := testutil.NewTestStore(t) + src, err := st.GetOrCreateSource("gmail", "index-reject@example.com") + require.NoError(err, "GetOrCreateSource") + + require.NoError(st.AddAccountIdentity(src.ID, "bob@example.com", "manual")) + + // A direct SQL writer supplying the derived key for a case variant must + // hit the partial unique index: this is the schema-level enforcement the + // application-level compare could not provide. + _, err = st.DB().Exec(st.Rebind( + `INSERT INTO account_identities (source_id, address, address_key, source_signal) + VALUES (?, ?, ?, ?)`), + src.ID, "Bob@Example.com", "bob@example.com", "raw") + require.Error(err, "keyed case-variant duplicate must violate the unique index") +} + +func TestAccountIdentityLegacyOmittedKeyInsertSucceedsAndIsRepaired(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + st := testutil.NewTestStore(t) + src, err := st.GetOrCreateSource("gmail", "legacy-writer@example.com") + require.NoError(err, "GetOrCreateSource") + + require.NoError(st.AddAccountIdentity(src.ID, "Carol@Example.com", "manual")) + + // A previous-release binary inserts with its old column list. The '' + // default keeps the write working (the partial index exempts ''), even + // though a keyed row for the same logical identity already exists. + _, err = st.DB().Exec(st.Rebind( + `INSERT INTO account_identities (source_id, address, source_signal) + VALUES (?, ?, ?)`), + src.ID, "carol@example.com", "legacy") + require.NoError(err, "previous-release column list must keep working") + + require.NoError(st.InitSchema(), "reinit schema to run the key repair") + + identities, err := st.ListAccountIdentities(src.ID) + require.NoError(err, "ListAccountIdentities") + require.Len(identities, 1, "legacy duplicate must merge into the keyed row") + assert.Equal("Carol@Example.com", identities[0].Address, + "the row already carrying the correct key survives") + assert.Equal("legacy,manual", identities[0].SourceSignal, "signal sets must union") +} + +func TestAddAccountIdentityCaseVariantsMergeToOneRow(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + st := testutil.NewTestStore(t) + src, err := st.GetOrCreateSource("gmail", "add-case-variant@example.com") + require.NoError(err, "GetOrCreateSource") + + require.NoError(st.AddAccountIdentity(src.ID, "Dana@Example.com", "manual")) + require.NoError(st.AddAccountIdentity(src.ID, "dana@example.com", "header")) + + identities, err := st.ListAccountIdentities(src.ID) + require.NoError(err, "ListAccountIdentities") + require.Len(identities, 1, "case-variant confirmations must merge, not diverge") + assert.Equal("Dana@Example.com", identities[0].Address, + "display casing stays as first written") + assert.Equal("header,manual", identities[0].SourceSignal) + + // Case-aware removal still works after the keyed lookup change. + removed, err := st.RemoveAccountIdentity(src.ID, "DANA@EXAMPLE.COM") + require.NoError(err, "RemoveAccountIdentity") + assert.Equal(int64(1), removed) +} + +// TestAccountIdentityKeyRepairRunsAfterProvenanceInitialization proves the +// InitSchema ordering contract: on a pre-provenance archive, the duplicate +// collapse's attribution refresh must not run before source_is_from_me is +// initialized, or the provenance backfill would read identity-derived +// is_from_me values as source-native and bake them in permanently. +func TestAccountIdentityKeyRepairRunsAfterProvenanceInitialization(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + f := storetest.New(t) + st := f.Store + senderID := f.EnsureParticipant("owner2@example.com", "Owner Two", "example.com") + + src, err := st.GetOrCreateSource("gmail", "provenance-order@example.com") + require.NoError(err, "GetOrCreateSource") + convID, err := st.EnsureConversation(src.ID, "provenance-order-conversation", "Thread") + require.NoError(err, "EnsureConversation") + messageID, err := st.UpsertMessage(&store.Message{ + SourceID: src.ID, + ConversationID: convID, + SourceMessageID: "provenance-order-message", + SenderID: sql.NullInt64{Int64: senderID, Valid: true}, + IsFromMe: false, + }) + require.NoError(err, "persist received message") + + // Case-variant duplicate identities written before the key column, plus + // a message predating attribution provenance: the archive state an + // upgrade encounters. + for _, addr := range []string{"Owner2@Example.com", "owner2@example.com"} { + _, err = st.DB().Exec(st.Rebind(` + INSERT INTO account_identities (source_id, address, source_signal) + VALUES (?, ?, ?)`), src.ID, addr, "manual") + require.NoError(err, "seed legacy identity %s", addr) + } + _, err = st.DB().Exec(st.Rebind(` + UPDATE messages + SET source_is_from_me = NULL, identity_is_from_me = FALSE + WHERE id = ?`), messageID) + require.NoError(err, "simulate pre-provenance message row") + _, err = st.DB().Exec(st.Rebind(` + DELETE FROM applied_migrations WHERE name = ?`), + "message_attribution_provenance_v3") + require.NoError(err, "reset attribution migration sentinel") + + require.NoError(st.InitSchema(), "run production schema migration") + + identities, err := st.ListAccountIdentities(src.ID) + require.NoError(err, "ListAccountIdentities") + require.Len(identities, 1, "duplicates must collapse during the same InitSchema") + + var sourceDerived sql.NullBool + var identityDerived bool + require.NoError(st.DB().QueryRow(st.Rebind(` + SELECT source_is_from_me, identity_is_from_me + FROM messages WHERE id = ?`), messageID).Scan(&sourceDerived, &identityDerived)) + assert.True(sourceDerived.Valid, "provenance must be initialized") + assert.False(sourceDerived.Bool, + "identity-derived attribution must not be recorded as source-native") + assert.True(identityDerived, "sender matches a confirmed identity") + isFromMe, err := st.GetMessageIsFromMe(messageID) + require.NoError(err, "GetMessageIsFromMe") + assert.True(isFromMe, "effective attribution comes from the identity match") +} diff --git a/internal/store/migrate_legacy_identity.go b/internal/store/migrate_legacy_identity.go index e00ea58e0..db368b4c1 100644 --- a/internal/store/migrate_legacy_identity.go +++ b/internal/store/migrate_legacy_identity.go @@ -150,46 +150,23 @@ func (s *Store) MigrateLegacyIdentityConfigContext( for _, addr := range normalized { // Comparison rule (email-shaped → case-insensitive; // everything else → case-sensitive) is shared with - // AddAccountIdentity via identifierMatch — see - // identifier_match.go. - match := newIdentifierMatch(addr) - var existing string - qerr := tx.QueryRowContext(ctx, - `SELECT source_signal FROM account_identities - WHERE source_id = ? AND `+match.WhereClause("address"), - src.ID, match.BindValue(), - ).Scan(&existing) - switch { - case errors.Is(qerr, sql.ErrNoRows): - _, txErr := tx.ExecContext(ctx, - `INSERT INTO account_identities (source_id, address, source_signal) - VALUES (?, ?, ?)`, - src.ID, addr, "config_migration", - ) - if txErr != nil { - return fmt.Errorf("insert identity (source=%d, addr=%s): %w", src.ID, addr, txErr) - } - // A brand new (source_id, address) pair changes - // owner_participants and the is_from_me derivation for - // this source, exactly like AddAccountIdentity's insert - // branch — see the matching comment there. + // AddAccountIdentity via the row-level merge helper, + // which keys on the persisted address_key — see + // mergeAccountIdentitySignalsTxWith. + inserted, merr := s.mergeAccountIdentitySignalsTx( + ctx, tx, src.ID, addr, + []string{"config_migration"}, newIdentifierMatch(addr), + ) + if merr != nil { + return fmt.Errorf("merge identity (source=%d, addr=%s): %w", src.ID, addr, merr) + } + if inserted { + // A brand new identity row changes owner_participants + // and the is_from_me derivation for this source, + // exactly like AddAccountIdentity's insert branch — + // see the matching comment there. insertedAny = true insertedForSource = true - case qerr != nil: - return fmt.Errorf("read existing identity (source=%d, addr=%s): %w", src.ID, addr, qerr) - default: - merged := mergeSignalSet(existing, "config_migration") - if merged != existing { - _, uerr := tx.ExecContext(ctx, - `UPDATE account_identities - SET source_signal = ? - WHERE source_id = ? AND `+match.WhereClause("address"), - merged, src.ID, match.BindValue(), - ) - if uerr != nil { - return fmt.Errorf("update identity (source=%d, addr=%s): %w", src.ID, addr, uerr) - } - } } } if insertedForSource { diff --git a/internal/store/migrations.go b/internal/store/migrations.go index 9f84d3e69..043f2d01e 100644 --- a/internal/store/migrations.go +++ b/internal/store/migrations.go @@ -75,6 +75,11 @@ const ( migrationActivityProjectionTriggers = "activity_projection_triggers_v4" migrationPersonInferenceProviderV2 = "person_inference_provider_v2" migrationPersonSweepCallsV2 = "person_sweep_calls_v2" + // The partial unique index on account_identities(source_id, address_key) + // is DDL that runs once per archive; the key backfill itself is not + // ledgered because previous-release writers can reintroduce unkeyed rows + // at any time (see ensureAccountIdentityAddressKeys). + migrationAccountIdentityAddressKeyIndex = "account_identities_address_key_index_v1" ) func (s *Store) classifyLegacyGmailChats(ctx context.Context, tx *loggedTx) error { diff --git a/internal/store/schema.sql b/internal/store/schema.sql index 324543cce..fefa2e749 100644 --- a/internal/store/schema.sql +++ b/internal/store/schema.sql @@ -1784,9 +1784,19 @@ CREATE TABLE IF NOT EXISTS saved_views ( -- Confirmed per-account "me" identities used by sent-message detection -- in dedup. Identity is account-scoped: an address confirmed for one -- source does not imply it is "me" in any other source. +-- address_key is NormalizeIdentifierForCompare(address): lowercased for +-- email-shaped identifiers, verbatim otherwise. It is the comparison- +-- canonical form the application already matches on; persisting it lets a +-- partial unique index enforce one row per logical identity at the schema +-- level (see ensureAccountIdentityAddressKeys). The '' default is a +-- sentinel for rows written by binaries that predate the column; the next +-- store open derives their keys and merges any case-variant duplicates. +-- The index itself is created in Go after the legacy-column migrations, +-- because on upgraded archives this file runs before the column exists. CREATE TABLE IF NOT EXISTS account_identities ( source_id INTEGER NOT NULL REFERENCES sources(id) ON DELETE CASCADE, address TEXT NOT NULL, -- case-preserved + address_key TEXT NOT NULL DEFAULT '', -- comparison key; '' = needs derivation source_signal TEXT NOT NULL DEFAULT '', -- sorted comma-separated signal set, e.g. 'manual' or 'account-identifier,manual' confirmed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (source_id, address) diff --git a/internal/store/schema_pg.sql b/internal/store/schema_pg.sql index aff850366..cb2076c08 100644 --- a/internal/store/schema_pg.sql +++ b/internal/store/schema_pg.sql @@ -1797,9 +1797,14 @@ CREATE TABLE IF NOT EXISTS saved_views ( -- Confirmed per-account "me" identities used by sent-message detection -- in dedup. Identity is account-scoped: an address confirmed for one -- source does not imply it is "me" in any other source. +-- address_key mirrors the SQLite schema: the comparison-canonical form of +-- address, '' for rows written by binaries that predate the column. The +-- partial unique index on (source_id, address_key) is created in Go after +-- the legacy-column migrations (see ensureAccountIdentityAddressKeys). CREATE TABLE IF NOT EXISTS account_identities ( source_id BIGINT NOT NULL REFERENCES sources(id) ON DELETE CASCADE, address TEXT NOT NULL, + address_key TEXT NOT NULL DEFAULT '', source_signal TEXT NOT NULL DEFAULT '', confirmed_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (source_id, address) diff --git a/internal/store/store.go b/internal/store/store.go index c67045a66..04c4aaa1d 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -1535,6 +1535,17 @@ func (s *Store) InitSchemaContext(ctx context.Context) error { return err } + // Runs after the legacy-column loop (upgraded archives need the + // address_key column before it is read, backfilled, and uniquely + // indexed) and after the attribution-provenance migration above: a + // duplicate collapse refreshes message attribution, and that refresh + // folds identity matches into is_from_me. On a pre-provenance archive + // the backfill would then read those identity-derived values as + // source-native, permanently mislabeling ownership provenance. + if err := s.ensureAccountIdentityAddressKeys(ctx); err != nil { + return fmt.Errorf("ensure account identity address keys: %w", err) + } + // Identity discovery scans one source in message-ID order. On SQLite the // plain idx_messages_source index already orders ties by rowid, so no // separate composite index is needed there (see schema.sql). PostgreSQL