[Large-room policy 5/5] feat(chat): keep cold rooms out of people features

cmd/msgvault/cmd/build_cache.go

@@ -203,6 +203,11 @@ var buildCacheBeforeMessagesExportHook func() error
 // Tests override it to exercise platform-specific snapshot behavior.
 var cacheSnapshotGOOS = runtime.GOOS
 
+const (
+	cacheExportParticipants           = "cache_export_participants"
+	cacheExportParticipantIdentifiers = "cache_export_participant_identifiers"
+)
+
 // buildCacheWriteStateFile atomically publishes the cache sync state marker;
 // a test seam for simulating state persistence failures.
 var buildCacheWriteStateFile = writeCacheStateFile
@@ -927,7 +932,7 @@ func (s *cacheSourceSnapshot) participantIdentifiersExportSelectSQL() string {
 			COALESCE(%s, '') AS identifier_value,
 			COALESCE(%s, '') AS display_value,
 			COALESCE(TRY_CAST(is_primary AS BOOLEAN), false) AS is_primary
-		FROM sqlite_db.participant_identifiers`, s.identityExportSQL("identifier_type"), s.identityExportSQL("identifier_value"), s.textSQL("display_value"))
+		FROM %s pi`, s.participantIdentityExportSQL("pi.identifier_type"), s.participantIdentityExportSQL("pi.identifier_value"), s.participantTextSQL("pi.display_value"), s.snapshotTable(cacheExportParticipantIdentifiers))
 }
 
 // participantsExportSelectSQL renders the participants dataset export. The
@@ -937,12 +942,12 @@ func (s *cacheSourceSnapshot) participantIdentifiersExportSelectSQL() string {
 // (unknown), not as a repaired value that could collide with another address.
 func (s *cacheSourceSnapshot) participantsExportSelectSQL() string {
 	return fmt.Sprintf(`SELECT
-			id,
+			p.id,
 			COALESCE(%s, '') AS email_address,
 			COALESCE(%s, '') AS domain,
 			COALESCE(%s, '') AS display_name,
 			COALESCE(%s, '') AS phone_number
-		FROM sqlite_db.participants`, s.identityExportSQL("email_address"), s.textSQL("domain"), s.textSQL("display_name"), s.textSQL("phone_number"))
+		FROM %s p`, s.participantIdentityExportSQL("p.email_address"), s.participantTextSQL("p.domain"), s.participantTextSQL("p.display_name"), s.participantTextSQL("p.phone_number"), s.snapshotTable(cacheExportParticipants))
 }
 
 // personDisplayNamesExportSelectSQL keeps full and derived-only exports identical.
@@ -2013,6 +2018,13 @@ func (s *cacheSourceSnapshot) DuckDB() sqlRunner {
 	return s.duckDB
 }
 
+func (s *cacheSourceSnapshot) snapshotTable(name string) string {
+	if s.csvSnapshot {
+		return "sqlite_db." + name
+	}
+	return name
+}
+
 // identityPresenceSQL preserves raw address presence independently of UTF-8
 // validity. CSV snapshots carry a boolean computed by SQLite before invalid
 // identity bytes become NULL; scanner snapshots still have the raw bytes.
@@ -2024,16 +2036,23 @@ func (s *cacheSourceSnapshot) identityPresenceSQL(emailRef, presenceRef string)
 }
 
 // Prepare materializes the CSV fallback after metadata has pinned the SQLite
-// read transaction. sqlite_scanner needs no preparation because DuckDB's
-// transaction reads the attached database directly.
+// read transaction. The sqlite_scanner path also prepares the participant
+// export projections through sqlite_query in that same snapshot.
 func (s *cacheSourceSnapshot) Prepare(through int64) error {
-	return s.prepareTables(s.tables(through))
+	if s.csvSnapshot {
+		return s.prepareTables(s.tables(through))
+	}
+	return s.prepareParticipantExportTables(through)
 }
 
 // PrepareDatasets materializes only the named SQLite tables on the CSV
-// fallback path. sqlite_scanner needs no materialization. Derived refreshes
-// use this to avoid exporting unrelated multi-million-row junction tables.
+// fallback path. sqlite_scanner reads them directly and materializes only the
+// participant export projections. Derived refreshes use this to avoid
+// exporting unrelated multi-million-row junction tables.
 func (s *cacheSourceSnapshot) PrepareDatasets(through int64, names ...string) error {
+	if !s.csvSnapshot {
+		return s.prepareParticipantExportTables(through)
+	}
 	wanted := make(map[string]bool, len(names))
 	for _, name := range names {
 		wanted[name] = true
@@ -2056,6 +2075,32 @@ func (s *cacheSourceSnapshot) PrepareDatasets(through int64, names ...string) er
 	return s.prepareTables(selected)
 }
 
+// prepareParticipantExportTables evaluates value-level policy in SQLite,
+// where the store's eligibility predicates have native query planning. The
+// DuckDB projection then reads only the resulting safe participant values.
+func (s *cacheSourceSnapshot) prepareParticipantExportTables(through int64) error {
+	if s.duckTx == nil {
+		return errors.New("prepare participant exports: missing DuckDB snapshot transaction")
+	}
+	for _, table := range s.tables(through) {
+		if table.name != cacheExportParticipants && table.name != cacheExportParticipantIdentifiers {
+			continue
+		}
+		var querySQL string
+		if table.name == cacheExportParticipants {
+			querySQL = participantExportQuery(true)
+		} else {
+			querySQL = participantIdentifierExportQuery(true)
+		}
+		query := strings.ReplaceAll(querySQL, "'", "''")
+		statement := fmt.Sprintf(`CREATE TEMP TABLE "%s" AS SELECT * FROM sqlite_query('sqlite_db', '%s')`, table.name, query)
+		if _, err := s.duckTx.Exec(statement); err != nil {
+			return fmt.Errorf("prepare %s snapshot: %w", table.name, err)
+		}
+	}
+	return nil
+}
+
 func (s *cacheSourceSnapshot) tables(through int64) []cacheSnapshotTable {
 	attachmentMIMEColumn := "'' AS mime_type"
 	if s.hasAttachmentMIME {
@@ -2093,6 +2138,8 @@ func (s *cacheSourceSnapshot) tables(through int64) []cacheSnapshotTable {
 		messageTypes += ", 'account_address': 'VARCHAR', 'account_path': 'VARCHAR'"
 	}
 	messageTypes += "}"
+	participantExportQuery := participantExportQuery(false)
+	participantIdentifierExportQuery := participantIdentifierExportQuery(false)
 
 	// Every column carries an explicit type so the schema DuckDB sees never
 	// depends on the data: read_csv_auto sniffs empty or all-NULL columns as
@@ -2116,14 +2163,24 @@ func (s *cacheSourceSnapshot) tables(through int64) []cacheSnapshotTable {
 			"types={'id': 'BIGINT', 'message_id': 'BIGINT', 'size': 'BIGINT', 'filename': 'VARCHAR', 'mime_type': 'VARCHAR', 'attachment_metadata': 'VARCHAR'}", nil},
 		{"persons", "SELECT id, display_name FROM persons", "types={'id': 'BIGINT', 'display_name': 'VARCHAR'}", nil},
 		{"person_participants", "SELECT person_id, participant_id FROM person_participants", "types={'person_id': 'BIGINT', 'participant_id': 'BIGINT'}", nil},
+		{"person_contact_points", "SELECT person_id,address_kind,normalized_value,source,active_until,superseded_at FROM person_contact_points", "types={'person_id':'BIGINT','address_kind':'VARCHAR','normalized_value':'VARCHAR','source':'VARCHAR','active_until':'TIMESTAMP','superseded_at':'TIMESTAMP'}", identityColumns("normalized_value")},
+		{"participant_contact_observations", "SELECT id,participant_id,source_id,address_kind,normalized_value,source,active_until,superseded_at FROM participant_contact_observations", "types={'id':'BIGINT','participant_id':'BIGINT','source_id':'BIGINT','address_kind':'VARCHAR','normalized_value':'VARCHAR','source':'VARCHAR','active_until':'TIMESTAMP','superseded_at':'TIMESTAMP'}", identityColumns("normalized_value")},
+		{"participant_contact_observation_conversations", "SELECT observation_id,conversation_id FROM participant_contact_observation_conversations", "types={'observation_id':'BIGINT','conversation_id':'BIGINT'}", nil},
+		{"communication_services", "SELECT id,slug FROM communication_services", "types={'id':'BIGINT','slug':'VARCHAR'}", identityColumns("slug")},
 		{tableParticipants, "SELECT id, email_address, domain, display_name, phone_number, " + emailPresence + " AS primary_email_present FROM participants",
 			"types={'id': 'BIGINT', 'email_address': 'VARCHAR', 'domain': 'VARCHAR', 'display_name': 'VARCHAR', 'phone_number': 'VARCHAR', 'primary_email_present': 'BOOLEAN'}",
 			identityColumns("email_address")},
 		{"account_identities", "SELECT source_id, address FROM account_identities",
 			"types={'source_id': 'BIGINT', 'address': 'VARCHAR'}",
 			identityColumns("address")},
-		{tableParticipantIdentifiers, "SELECT participant_id, identifier_type, identifier_value, display_value, is_primary FROM participant_identifiers",
-			"types={'participant_id': 'BIGINT', 'identifier_type': 'VARCHAR', 'identifier_value': 'VARCHAR', 'display_value': 'VARCHAR', 'is_primary': 'BOOLEAN'}",
+		{tableParticipantIdentifiers, "SELECT id, participant_id, identifier_type, identifier_value, display_value, is_primary, service_id, scope_kind, scope_value FROM participant_identifiers",
+			"types={'id':'BIGINT','participant_id': 'BIGINT', 'identifier_type': 'VARCHAR', 'identifier_value': 'VARCHAR', 'display_value': 'VARCHAR', 'is_primary': 'BOOLEAN','service_id':'BIGINT','scope_kind':'VARCHAR','scope_value':'VARCHAR'}",
+			identityColumns("identifier_type", "identifier_value")},
+		{cacheExportParticipants, participantExportQuery,
+			"types={'id':'BIGINT','email_address':'VARCHAR','domain':'VARCHAR','display_name':'VARCHAR','phone_number':'VARCHAR'}",
+			identityColumns("email_address")},
+		{cacheExportParticipantIdentifiers, participantIdentifierExportQuery,
+			"types={'participant_id':'BIGINT','identifier_type':'VARCHAR','identifier_value':'VARCHAR','display_value':'VARCHAR','is_primary':'BOOLEAN'}",
 			identityColumns("identifier_type", "identifier_value")},
 		{tableLabels, "SELECT id, name FROM labels",
 			"types={'id': 'BIGINT', 'name': 'VARCHAR'}", nil},
@@ -2136,6 +2193,42 @@ func (s *cacheSourceSnapshot) tables(through int64) []cacheSnapshotTable {
 	}
 }
 
+// participantExportQuery applies the value-level chat policy in SQLite. The
+// scanner path hex-encodes text so DuckDB never materializes malformed UTF-8
+// while creating the temporary table; the final cache projection decodes and
+// repairs display text or rejects invalid identity keys.
+func participantExportQuery(hexEncodeText bool) string {
+	value := func(eligible, column string) string {
+		expression := "CASE WHEN " + eligible + " THEN " + column + " END"
+		if hexEncodeText {
+			return "HEX(CAST(" + expression + " AS BLOB))"
+		}
+		return expression
+	}
+	return fmt.Sprintf(`SELECT p.id,
+		%s AS email_address,
+		%s AS domain,
+		%s AS display_name,
+		%s AS phone_number
+		FROM participants p`,
+		value(store.FullParticipantValueWhere("p.id", "'email'", "p.email_address"), "p.email_address"),
+		value(store.FullParticipantValueWhere("p.id", "'email'", "p.email_address"), "p.domain"),
+		value(store.FullParticipantNameWhere("p.id", "p.display_name"), "p.display_name"),
+		value(store.FullParticipantValueWhere("p.id", "'phone'", "p.phone_number"), "p.phone_number"),
+	)
+}
+
+func participantIdentifierExportQuery(hexEncodeText bool) string {
+	identifierType, identifierValue, displayValue := "pi.identifier_type", "pi.identifier_value", "pi.display_value"
+	if hexEncodeText {
+		identifierType = "HEX(CAST(" + identifierType + " AS BLOB))"
+		identifierValue = "HEX(CAST(" + identifierValue + " AS BLOB))"
+		displayValue = "HEX(CAST(" + displayValue + " AS BLOB))"
+	}
+	return `SELECT pi.participant_id, ` + identifierType + ` AS identifier_type, ` + identifierValue + ` AS identifier_value, ` + displayValue + ` AS display_value, pi.is_primary
+		FROM participant_identifiers pi WHERE ` + store.FullParticipantIdentifierWhereSQLite("pi")
+}
+
 func (s *cacheSourceSnapshot) prepareTables(tables []cacheSnapshotTable) error {
 	if s.sqliteTx == nil {
 		return nil

cmd/msgvault/cmd/build_cache_test.go

@@ -1016,6 +1016,68 @@ func TestBuildCache_PublishesConversationParticipants(t *testing.T) {
 	assert.Equal(t, int64(9), count)
 }
 
+func TestBuildCacheRedactsColdOnlyParticipantValuesFromExports(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		forceCSV bool
+	}{
+		{name: "sqlite scanner"},
+		{name: "CSV snapshot", forceCSV: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			require := require.New(t)
+			assert := assert.New(t)
+			if tc.forceCSV {
+				t.Setenv("MSGVAULT_FORCE_CSV_SNAPSHOT", "1")
+			} else {
+				t.Setenv("MSGVAULT_FORCE_CSV_SNAPSHOT", "")
+			}
+			tmpDir := setupTestSQLite(t)
+			dbPath := filepath.Join(tmpDir, "test.db")
+			db, err := sql.Open("sqlite3", dbPath)
+			require.NoError(err)
+			_, err = db.Exec(`
+				INSERT INTO sources (id,source_type,identifier) VALUES (2,'beeper','large-room-account');
+				INSERT INTO conversations (id,source_id,source_conversation_id,title,conversation_type) VALUES
+					(201,2,'full-room','Full room','group_chat'),
+					(202,2,'cold-room','Cold room','group_chat');
+				INSERT INTO conversation_policy (conversation_id,effective_tier) VALUES (201,'full'),(202,'cold');
+				INSERT INTO participants (id,email_address,domain,display_name,phone_number) VALUES
+					(5,'cold-only@example.test','example.test','Cold-only name','+15550109991');
+				INSERT INTO conversation_participants (conversation_id,participant_id) VALUES (201,5),(202,5);
+				INSERT INTO participant_identifiers
+					(id,participant_id,identifier_type,identifier_value,display_value,is_primary,service_id,scope_kind,scope_value)
+					VALUES (3,5,'beeper','beeper:account:9:cold-user','Cold-only user',1,NULL,NULL,NULL);
+			`)
+			require.NoError(err)
+			require.NoError(db.Close())
+
+			analyticsDir := filepath.Join(tmpDir, "analytics")
+			_, err = buildCache(dbPath, analyticsDir, false)
+			require.NoError(err)
+
+			duckdb, err := sql.Open("duckdb", "")
+			require.NoError(err)
+			defer func() { assert.NoError(duckdb.Close()) }()
+			var email, domain, name, phone string
+			err = duckdb.QueryRow(`SELECT email_address,domain,display_name,phone_number
+				FROM read_parquet(?) WHERE id=5`, filepath.Join(analyticsDir, "participants", "*.parquet"),
+			).Scan(&email, &domain, &name, &phone)
+			require.NoError(err)
+			assert.Empty(email)
+			assert.Empty(domain)
+			assert.Empty(name)
+			assert.Empty(phone)
+			var identifiers int64
+			err = duckdb.QueryRow(`SELECT COUNT(*) FROM read_parquet(?) WHERE participant_id=5`,
+				filepath.Join(analyticsDir, "participant_identifiers", "*.parquet"),
+			).Scan(&identifiers)
+			require.NoError(err)
+			assert.Zero(identifiers)
+		})
+	}
+}
+
 // TestBuildCache_ExportsRecipientEnvelopeAddress verifies both address
 // columns of the message_recipients Parquet dataset (cache schema v26).
 // envelope_address is the header address exactly as the store recorded it and

cmd/msgvault/cmd/cache_derived.go

@@ -173,12 +173,16 @@ func refreshDerivedDatasetsOnly(
 		"cache_unsent_messages",
 		"conversation_policy", "sources",
 		tableMessages,
+		"message_recipients",
 		tableConversations,
 		tableConversationParticipants,
 		"account_identities",
 		tableParticipants,
 		tableParticipantIdentifiers,
 		"persons", "person_participants",
+		"person_contact_points", "participant_contact_observations",
+		"participant_contact_observation_conversations", "communication_services",
+		cacheExportParticipants, cacheExportParticipantIdentifiers,
 	}
 	var relatedChangeSeq int64
 	var relatedKinds relatedChangeKinds
@@ -326,23 +330,28 @@ func refreshDerivedDatasetsOnly(
 			return nil, fmt.Errorf("export derived conversation participants: %w", err)
 		}
 	}
-	identifiersChanged :=
-		participantIdentifierRevision != state.ParticipantIdentifierRevision
-	if identifiersChanged {
+	identifierProjectionChanged :=
+		participantIdentifierRevision != state.ParticipantIdentifierRevision ||
+			(repairRelated && relatedKinds.recipients)
+	if identifierProjectionChanged {
 		// The directory rebuild reads identifier values from the
 		// participant_identifiers base dataset (relationship_people search
-		// values and label fallbacks), so a changed mapping must be re-staged
-		// and republished alongside the derived index.
+		// values and label fallbacks). Recipient history also controls native
+		// identifier eligibility, so a recipient-only repair must re-stage this
+		// projection even when the identifier mapping revision did not change.
 		if err := derivedCopy(tableParticipantIdentifiers, sourceSnapshot.participantIdentifiersExportSelectSQL()); err != nil {
 			return nil, fmt.Errorf("export derived participant identifiers: %w", err)
 		}
 	}
 	displayNamesChanged :=
 		participantDisplayNameRevision != state.ParticipantDisplayNameRevision
-	if identifiersChanged || displayNamesChanged {
+	participantProjectionChanged := identifierProjectionChanged || displayNamesChanged ||
+		(repairRelated && relatedKinds.recipients)
+	if participantProjectionChanged {
 		// Participant identifiers can create participant rows, and display-name
-		// mutations change the row already present in participants.parquet. Both
-		// changes must replace that base dataset before rebuilding the directory.
+		// mutations change the row already present in participants.parquet. Recipient
+		// evidence can also change value eligibility without advancing either
+		// revision, so those repairs must refresh the same base dataset.
 		if err := derivedCopy(tableParticipants, sourceSnapshot.participantsExportSelectSQL()); err != nil {
 			return nil, fmt.Errorf("export derived participants: %w", err)
 		}
@@ -411,8 +420,8 @@ func refreshDerivedDatasetsOnly(
 	plan := derivedCachePublishPlan(
 		conversationChanged,
 		typesChanged,
-		identifiersChanged,
-		identifiersChanged || displayNamesChanged,
+		identifierProjectionChanged,
+		participantProjectionChanged,
 		personDisplayNamesChanged || identityChanged,
 	)
 	if repairRelated {