fix(store): enforce case-folded uniqueness for account identities

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)
 			}

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"},

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"},

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]
+}

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)
+	}
+}