feat(relay): match verified names in profile search (#9019)

## Summary

`POST /profiles/search` only matches kind:0 `name` and `display_name`.
In communities that publish NIP-FI verified profiles, a person's surname
or directory username often appears only in their relay-signed
kind:30382, so typeahead cannot find them by it.

- Where the community advertises verified profiles, the search also
matches the `display_name`, `username` and `full_name` tags of each
current kind:30382 head signed by the relay key.
- Verified names need a trailing prefix of at least three characters, as
in NIP-50, so a short prefix doesn't make the relay rank every verified
head. kind:0 keeps one-character typeahead. Lookups from heads to
`events` also match `created_at`, so Postgres prunes to one partition.
- Results are paged by person. Each person returns their current kind:0,
when they have one, then their current kind:30382, when they have one,
whichever matched. A person is ranked by their best match across both
sources, using the existing tiers.
- NIP-11 `federated_identity.public_profile` gains `searchable: true`,
so clients can rely on the response instead of a separate kind:30382
read.
- Migration `0081` adds a GIN expression index over the head's name tags
on `nip_fi_profile_heads` (built-in `jsonb_path_query_first` and
`to_tsvector`, no new functions). The table holds one row per verified
person. Like `0078`, the migration builds the index only while the table
is empty; populated databases prebuild it with `CREATE INDEX
CONCURRENTLY` (`docs/profile-search.md`), and the migration then checks
it is valid and has the expected definition.

Compatibility: the request schema is unchanged. Existing clients that
keep only kind:0 events work as before and also find people who match
only by verified name. Trust stays pinned to the relay `self` key; the
event shape follows NIP-85 user assertions, but discovery is
relay-advertised rather than kind:10040.

## Testing

- `buzz-search` Postgres integration: new
`profile_names_match_verified_names` covers forged-key heads, replaced
heads, other communities, people without kind:0, per-person grouping,
paging, and words split across sources; the existing profile-name tests
pass unchanged. The test fails when the author filter is inverted.
- `buzz-relay` end to end in enforce mode:
`profile_search_matches_verified_names_where_advertised` publishes
through the real writer and finds the person by surname only where
verified profiles are advertised.
- Desired-state and migration index definitions match.
- `profile_names_use_verified_name_index` asserts the search scans the
new index. Migration `0081` checked against an empty table, a rerun, a
populated table without the index (raises), the documented concurrent
prebuild, and an index with a different definition (raises).
- Latency, 50k people each with a kind:0 and a verified head, release
build, local Postgres, best of 5, 30 people per page: 11–22 ms with
verified names versus 1–8 ms kind:0 only. `EXPLAIN` shows a bitmap scan
on the new index.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Signed-off-by: Kalvin Chau <kalvin@block.xyz>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

crates/buzz-db/src/runtime/migration.rs

@@ -777,7 +777,7 @@ mod postgres_tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 80);
+        assert_eq!(migrations.len(), 81);
         for (index, migration) in migrations.iter().enumerate() {
             assert_eq!(migration.version, (index + 1) as i64);
         }

crates/buzz-relay/src/api/bridge.rs

@@ -7176,6 +7176,103 @@ pub(crate) mod postgres_tests {
         );
     }
 
+    // ── Profile name search — verified names, end to end ───────────────────
+    //
+    // Where the community advertises verified profiles, a member finds a
+    // person by a name only their relay-signed kind:30382 carries, and gets
+    // their kind:0 and that kind:30382. Without the advertisement, the same
+    // query finds nobody.
+    #[test]
+    #[ignore = "requires Postgres"]
+    fn profile_search_matches_verified_names_where_advertised() {
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("current_thread runtime");
+
+        let Some(unadvertised) = rt.block_on(nip_fi_enforce_test_state_with_verifier()) else {
+            panic!("local Postgres not reachable");
+        };
+        let mut advertised = (*unadvertised).clone();
+        let mut config = (*advertised.config).clone();
+        config.relay_private_key = Some(advertised.relay_keypair.secret_key().to_secret_hex());
+        config.nip_fi.profile_policies = vec![buzz_db::verified_profiles::ProfilePolicy {
+            issuer: "https://issuer.test".to_owned(),
+            label: Some("Example".to_owned()),
+            refresh_seconds: 3600,
+        }];
+        advertised.config = Arc::new(config);
+        let advertised = Arc::new(advertised);
+
+        let host = format!(
+            "nip-fi-verified-search-{}.local",
+            uuid::Uuid::new_v4().simple()
+        );
+        let community = rt
+            .block_on(advertised.db.ensure_configured_community(&host))
+            .expect("ensure community")
+            .id;
+        let subject = Keys::generate();
+        let profile = nostr::EventBuilder::new(nostr::Kind::Metadata, r#"{"name":"Fake"}"#)
+            .sign_with_keys(&subject)
+            .expect("sign profile");
+        rt.block_on(
+            advertised
+                .db
+                .replace_addressable_event(community, &profile, None),
+        )
+        .expect("store profile");
+        let now = chrono::Utc::now();
+        let verified = rt
+            .block_on(advertised.db.publish_verified_profile(
+                buzz_db::verified_profiles::ProfileAdmission {
+                    community,
+                    issuer: "https://issuer.test",
+                    pubkey: &subject.public_key().to_hex(),
+                    issued_at: now,
+                    admitted_at: now,
+                    display_name: "fname",
+                    username: "fname",
+                    full_name: Some("Fake Name"),
+                    label: "Example",
+                    refresh_seconds: 3600,
+                    membership_required: false,
+                },
+                &advertised.relay_keypair,
+            ))
+            .expect("publish verified profile")
+            .expect("verified profile written");
+
+        let search = |state: &Arc<crate::state::AppState>| {
+            let keys = Keys::generate();
+            let body = br#"{"query":"name"}"#;
+            let url = format!("https://{host}{}", crate::api::profiles::SEARCH_PATH);
+            let headers = same_key_nip98_and_assertion_headers(&keys, &url, "POST", body);
+            let (status, _resp_headers, resp) = rt.block_on(oneshot_request_full(
+                Arc::clone(state),
+                "POST",
+                crate::api::profiles::SEARCH_PATH,
+                &host,
+                headers,
+                body,
+            ));
+            assert_eq!(
+                status,
+                axum::http::StatusCode::OK,
+                "body: {:?}",
+                String::from_utf8_lossy(&resp)
+            );
+            serde_json::from_slice::<Vec<nostr::Event>>(&resp)
+                .expect("response is a JSON array of events")
+        };
+
+        let events = search(&advertised);
+        assert_eq!(events, vec![profile, verified]);
+        assert_eq!(events[1].pubkey, advertised.relay_keypair.public_key());
+        assert!(events.iter().all(|event| event.verify().is_ok()));
+        assert!(search(&unadvertised).is_empty());
+    }
+
     // ── Moderation reports — Enforce mode, same-key admission → reaches handler ─
     //
     // Positive control: a valid NIP-FI assertion + same-key NIP-98 on the

crates/buzz-relay/src/api/profiles.rs

@@ -3,8 +3,10 @@
 //! NIP-50 search over kind:0 matches the full profile JSON, so text inside
 //! avatar data URLs, `about` and `nip05` competes with real names. This
 //! endpoint keeps NIP-50 prefix matching and paging but applies them only to
-//! the `name` and `display_name` fields of each current profile, and returns
-//! the signed kind:0 events.
+//! the `name` and `display_name` fields of each current profile and, where
+//! the community advertises verified profiles, to the names in each current
+//! relay-signed kind:30382. Pages count people; each person returns their
+//! signed kind:0 and, when they have one, their kind:30382.
 
 use std::{collections::HashMap, sync::Arc};
 
@@ -72,8 +74,9 @@ fn parse_request(body: &[u8]) -> Result<Search, (StatusCode, Json<Value>)> {
     Ok((query.to_owned(), limit, page))
 }
 
-/// Search current profiles by `name` and `display_name` for an authenticated
-/// member. Responds with one page of signed kind:0 events, best match first.
+/// Search people by profile and verified names for an authenticated member.
+/// Responds with one page of signed kind:0 and kind:30382 events, best match
+/// first.
 pub async fn search(
     State(state): State<Arc<AppState>>,
     slot: crate::access::AccessSlot,
@@ -98,9 +101,13 @@ async fn search_inner(
         bridge::authenticate_member_post(&state, &slot, &headers, SEARCH_PATH, &body).await?;
     let (query, limit, page) = parse_request(&body).map_err(IntoResponse::into_response)?;
 
+    // Verified profiles count only where this community advertises them.
+    let verified_by = crate::verified_profiles::discovery(&state, &tenant)
+        .await
+        .map(|_| state.relay_keypair.public_key().to_bytes());
     let hits = state
         .search
-        .search_profile_names(tenant.community(), &query, page, limit)
+        .search_profile_names(tenant.community(), &query, verified_by, page, limit)
         .await
         .map_err(|e| search_read_error("profile search error", &e).into_response())?
         .hits;

crates/buzz-relay/src/verified_profiles.rs

@@ -139,5 +139,8 @@ pub(crate) async fn discovery(
             count += 1;
         }
     }
-    (count > 0).then(|| serde_json::json!({"kind": 30382, "multiple_issuers": count > 1}))
+    // `searchable`: `POST /profiles/search` also matches and returns these.
+    (count > 0).then(
+        || serde_json::json!({"kind": 30382, "multiple_issuers": count > 1, "searchable": true}),
+    )
 }

crates/buzz-search/src/lib.rs

@@ -69,15 +69,16 @@ impl SearchService {
         query::search(&self.pool, query).await
     }
 
-    /// Find current profiles by `name` and `display_name`; see
-    /// [`search_profile_names`].
+    /// Find people by profile name, and by verified profile names signed by
+    /// `verified_by`; see [`search_profile_names`].
     pub async fn search_profile_names(
         &self,
         community: CommunityId,
         q: &str,
+        verified_by: Option<[u8; 32]>,
         page: u32,
         per_page: u32,
     ) -> Result<SearchResult, SearchError> {
-        query::search_profile_names(&self.pool, community, q, page, per_page).await
+        query::search_profile_names(&self.pool, community, q, verified_by, page, per_page).await
     }
 }

crates/buzz-search/src/query.rs

@@ -442,27 +442,35 @@ fn tier_name(column: &str) -> String {
 /// Largest page size [`search_profile_names`] accepts.
 pub const PROFILE_NAMES_PER_PAGE_MAX: u32 = 100;
 
-/// Find current kind:0 profiles whose `name` or `display_name` matches `q`.
+/// Find people whose current kind:0 `name` or `display_name` matches `q`,
+/// or, with `verified_by`, whose current verified profile does.
 ///
 /// Matching is the NIP-50 [`SearchMode::Prefix`] rule (every completed query
 /// word must equal a word, the last may be a prefix, in any order), applied to
-/// the two name fields only. Other profile fields, such as `picture` data URLs,
-/// `about` and `nip05`, cannot satisfy a word. The result set is therefore the
-/// prefix-mode kind:0 result set minus profiles that match only outside their
-/// names; the existing `search_tsv` index stays the access path.
+/// name fields only. For kind:0 these are `name` and `display_name`; other
+/// profile fields, such as `picture` data URLs, `about` and `nip05`, cannot
+/// satisfy a word, and the existing `search_tsv` index stays the access path.
+/// With `verified_by`, a current kind:30382 event signed by that key matches
+/// on its `display_name`, `username` and `full_name` tags; its `d` tag names
+/// the person.
 ///
-/// Order: a name equal to the query, then a name that starts with it, then a
-/// name with a word that starts with it, then any other match. The query and
-/// names are compared lower-cased with whitespace collapsed, both normalized
-/// by the same SQL. Ties go by relevance of the query to the two names
-/// (`ts_rank_cd` over the names alone, not the whole profile), then newest
-/// first, then event id, so pages are stable. Pages are 1-based, like
-/// [`SearchQuery::page`].
+/// Order is per person, by their best match: a name equal to the query, then
+/// a name that starts with it, then a name with a word that starts with it,
+/// then any other match. The query and names are compared lower-cased with
+/// whitespace collapsed, both normalized by the same SQL. Ties go by
+/// relevance of the query to the names (`ts_rank_cd` over the names alone,
+/// not the whole event), then newest first, then pubkey, so pages are stable.
+/// Pages are 1-based, like [`SearchQuery::page`], and count people.
+///
+/// Each person yields their current kind:0, when they have one, then, with
+/// `verified_by`, their current verified profile, when they have one,
+/// whichever of the two matched.
 #[datastore_span(name = "search_profile_names", system = "postgresql")]
 pub async fn search_profile_names(
     pool: &PgPool,
     community: CommunityId,
     q: &str,
+    verified_by: Option<[u8; 32]>,
     page: u32,
     per_page: u32,
 ) -> Result<SearchResult, SearchError> {
@@ -475,54 +483,104 @@ pub async fn search_profile_names(
     };
     let per_page = per_page.clamp(1, PROFILE_NAMES_PER_PAGE_MAX);
     let offset = ((page - 1) as i64) * (per_page as i64);
-    // Tiers compare the query and each name after the same SQL normalization
-    // ([`TIER_TEXT`]), so the two sides agree by construction.
-    let mut qb: QueryBuilder<sqlx::Postgres> = QueryBuilder::new(
-        "SELECT id, kind, pubkey, channel_id, created_at_s, rank FROM (\
-         SELECT e.id, e.kind, e.pubkey, e.channel_id, \
-         EXTRACT(EPOCH FROM e.created_at)::bigint AS created_at_s, e.created_at, \
-         ts_rank_cd(names.tsv, search_query.query) AS rank, \
-         LEAST(",
-    );
-    for field in ["names.name", "names.display_name"] {
-        qb.push(format!(
-            "CASE WHEN {field} = search_query.typed THEN 0 \
-             WHEN starts_with({field}, search_query.typed) THEN 1 \
-             WHEN strpos({field}, ' ' || search_query.typed) > 0 THEN 2 ELSE 3 END"
-        ));
-        if field == "names.name" {
-            qb.push(", ");
-        }
-    }
-    qb.push(
-        ") AS tier \
-         FROM events e CROSS JOIN LATERAL (SELECT ",
-    );
+    let mut qb: QueryBuilder<sqlx::Postgres> = QueryBuilder::new("WITH search_query AS (SELECT ");
     push_tsquery(&mut qb, SearchMode::Prefix, &search_text, 0);
     qb.push(" AS query, ");
+    // Every verified match is ranked before `LIMIT`, so verified names need
+    // the NIP-50 minimum prefix; kind:0 keeps one-character typeahead.
+    push_tsquery(&mut qb, SearchMode::Prefix, &search_text, PREFIX_MIN_CHARS);
+    qb.push(" AS verified_query, ");
     qb.push(TIER_TEXT[0]);
     qb.push_bind(search_text.clone());
     qb.push(TIER_TEXT[1]);
-    qb.push(format!(
-        " AS typed) AS search_query \
-         CROSS JOIN LATERAL profile_name_fields(e.content) AS fields \
-         CROSS JOIN LATERAL (SELECT \
-           to_tsvector('simple', concat_ws(' ', fields.name, fields.display_name)) AS tsv, \
-           {} AS name, {} AS display_name) AS names \
-         WHERE e.community_id = ",
-        tier_name("fields.name"),
-        tier_name("fields.display_name"),
-    ));
+    qb.push(" AS typed), matches AS (SELECT e.pubkey AS person, ");
+    push_name_match(&mut qb, &["name", "display_name"], "e.created_at");
+    qb.push(
+        " FROM events e CROSS JOIN search_query \
+         CROSS JOIN LATERAL profile_name_fields(e.content) AS fields",
+    );
+    push_name_columns(&mut qb, &["name", "display_name"]);
+    qb.push(" WHERE e.community_id = ");
     qb.push_bind(*community.as_uuid());
     qb.push(
         " AND e.kind = 0 AND e.channel_id IS NULL AND e.deleted_at IS NULL \
          AND e.search_tsv @@ search_query.query \
-         AND names.tsv @@ search_query.query\
-         ) AS matched ORDER BY tier, rank DESC, created_at DESC, id LIMIT ",
+         AND names.tsv @@ search_query.query",
+    );
+    if let Some(author) = verified_by {
+        qb.push(" UNION ALL SELECT decode(pubkey, 'hex'), ");
+        push_name_match(
+            &mut qb,
+            &VERIFIED_NAME_TAGS,
+            "to_timestamp((event ->> 'created_at')::bigint)",
+        );
+        qb.push(" FROM nip_fi_profile_heads CROSS JOIN search_query CROSS JOIN LATERAL (SELECT ");
+        for (i, tag) in VERIFIED_NAME_TAGS.iter().enumerate() {
+            if i > 0 {
+                qb.push(", ");
+            }
+            qb.push(format!("{} AS {tag}", verified_name(tag)));
+        }
+        qb.push(") AS fields");
+        push_name_columns(&mut qb, &VERIFIED_NAME_TAGS);
+        qb.push(" WHERE community_id = ");
+        qb.push_bind(*community.as_uuid());
+        qb.push(format!(
+            " AND pubkey ~ '^[0-9a-f]{{64}}$' AND {} @@ search_query.verified_query",
+            verified_names_tsv()
+        ));
+        // The head counts only while its event is live and signed by `author`.
+        qb.push(" AND EXISTS (SELECT 1 FROM events e WHERE e.community_id = ");
+        qb.push_bind(*community.as_uuid());
+        // `created_at` lets Postgres prune `events` to the head's partition.
+        qb.push(
+            " AND e.id = decode(event ->> 'id', 'hex') \
+             AND e.created_at = to_timestamp((event ->> 'created_at')::bigint) \
+             AND e.kind = 30382 AND e.pubkey = ",
+        );
+        qb.push_bind(author.to_vec());
+        qb.push(" AND e.channel_id IS NULL AND e.deleted_at IS NULL)");
+    }
+    qb.push(
+        // Each person ranks by their single best match, not by mixing sources.
+        "), people AS (SELECT * FROM (SELECT DISTINCT ON (person) person, tier, \
+         rank, created_at FROM matches \
+         ORDER BY person, tier, rank DESC, created_at DESC) AS best \
+         ORDER BY tier, rank DESC, created_at DESC, person LIMIT ",
     );
     qb.push_bind(per_page as i64);
     qb.push(" OFFSET ");
     qb.push_bind(offset);
+    qb.push(
+        ") SELECT ev.id, ev.kind, ev.pubkey, ev.channel_id, \
+         EXTRACT(EPOCH FROM ev.created_at)::bigint AS created_at_s, people.rank \
+         FROM people CROSS JOIN LATERAL (\
+           (SELECT id, kind, pubkey, channel_id, created_at FROM events \
+            WHERE community_id = ",
+    );
+    qb.push_bind(*community.as_uuid());
+    qb.push(
+        " AND pubkey = people.person AND kind = 0 \
+         AND channel_id IS NULL AND deleted_at IS NULL \
+         ORDER BY created_at DESC, id LIMIT 1)",
+    );
+    if let Some(author) = verified_by {
+        qb.push(
+            " UNION ALL (SELECT e.id, e.kind, e.pubkey, e.channel_id, e.created_at \
+             FROM nip_fi_profile_heads h JOIN events e ON e.community_id = h.community_id \
+             AND e.id = decode(h.event ->> 'id', 'hex') \
+             AND e.created_at = to_timestamp((h.event ->> 'created_at')::bigint) \
+             WHERE h.community_id = ",
+        );
+        qb.push_bind(*community.as_uuid());
+        qb.push(" AND h.pubkey = encode(people.person, 'hex') AND e.kind = 30382 AND e.pubkey = ");
+        qb.push_bind(author.to_vec());
+        qb.push(" AND e.channel_id IS NULL AND e.deleted_at IS NULL)");
+    }
+    qb.push(
+        ") AS ev ORDER BY people.tier, people.rank DESC, people.created_at DESC, \
+         people.person, ev.kind",
+    );
 
     let rows = crate::cancel::fetch_all(pool, pool.acquire().await?, qb.build()).await?;
     let mut hits = Vec::with_capacity(rows.len());
@@ -547,6 +605,70 @@ pub async fn search_profile_names(
     Ok(SearchResult { hits, page })
 }
 
+/// Name tags of a relay-signed kind:30382 verified profile.
+const VERIFIED_NAME_TAGS: [&str; 3] = ["display_name", "username", "full_name"];
+
+/// The first value of `tag` in `nip_fi_profile_heads.event`. The relay
+/// writes each name tag at most once. Strict mode keeps a tag's array
+/// shape (lax mode unwraps it), and `silent` makes malformed tags no match.
+fn verified_name(tag: &str) -> String {
+    format!(
+        "(jsonb_path_query_first(event, 'strict $.tags[*] ? (@[0] == \"{tag}\")[1]', '{{}}', true) \
+         #>> '{{}}')"
+    )
+}
+
+/// The expression `idx_nip_fi_profile_heads_names` indexes
+/// (migrations/0081_nip_fi_profile_head_names.sql).
+fn verified_names_tsv() -> String {
+    let names = VERIFIED_NAME_TAGS
+        .iter()
+        .map(|tag| format!("COALESCE({}, '')", verified_name(tag)))
+        .collect::<Vec<_>>()
+        .join(" || ' ' || ");
+    format!("to_tsvector('simple', {names})")
+}
+
+/// Pushes `rank, created_at, tier` for one match over the `names` columns
+/// built by [`push_name_columns`] from the same `fields`.
+fn push_name_match(qb: &mut QueryBuilder<sqlx::Postgres>, fields: &[&str], created_at: &str) {
+    qb.push(format!(
+        "ts_rank_cd(names.tsv, search_query.query) AS rank, {created_at} AS created_at, LEAST("
+    ));
+    for (i, field) in fields.iter().enumerate() {
+        if i > 0 {
+            qb.push(", ");
+        }
+        let column = format!("names.{field}");
+        qb.push(format!(
+            "CASE WHEN {column} = search_query.typed THEN 0 \
+             WHEN starts_with({column}, search_query.typed) THEN 1 \
+             WHEN strpos({column}, ' ' || search_query.typed) > 0 THEN 2 ELSE 3 END"
+        ));
+    }
+    qb.push(") AS tier");
+}
+
+/// Pushes the `names` lateral: the `simple` tsvector of `fields` and each
+/// field normalized for tiers ([`TIER_TEXT`]).
+fn push_name_columns(qb: &mut QueryBuilder<sqlx::Postgres>, fields: &[&str]) {
+    let joined = fields
+        .iter()
+        .map(|field| format!("fields.{field}"))
+        .collect::<Vec<_>>()
+        .join(", ");
+    qb.push(format!(
+        " CROSS JOIN LATERAL (SELECT to_tsvector('simple', concat_ws(' ', {joined})) AS tsv"
+    ));
+    for field in fields {
+        qb.push(format!(
+            ", {} AS {field}",
+            tier_name(&format!("fields.{field}"))
+        ));
+    }
+    qb.push(") AS names");
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/buzz-search/tests/postgres_fts_integration.rs

@@ -1920,7 +1920,7 @@ async fn profile_names(
     per_page: u32,
 ) -> Vec<[u8; 32]> {
     SearchService::new(pool.clone())
-        .search_profile_names(community, q, page, per_page)
+        .search_profile_names(community, q, None, page, per_page)
         .await
         .expect("profile name search")
         .hits
@@ -2068,6 +2068,337 @@ async fn profile_names_survive_unparsable_content() {
     teardown(pool, &schema).await;
 }
 
+/// Insert a kind:30382 verified profile for `subject`, signed by `author`.
+async fn put_verified(
+    pool: &PgPool,
+    community: CommunityId,
+    author: [u8; 32],
+    subject: [u8; 32],
+    names: &[(&str, &str)],
+    at: i64,
+) -> [u8; 32] {
+    let id = rand_bytes32();
+    let d: String = subject.iter().map(|b| format!("{b:02x}")).collect();
+    let mut tags = vec![format!(r#"["d","{d}"]"#), format!(r#"["p","{d}"]"#)];
+    tags.extend(names.iter().map(|(k, v)| format!(r#"["{k}","{v}"]"#)));
+    let tags = format!("[{}]", tags.join(","));
+    sqlx::query(
+        "INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, d_tag) \
+         VALUES ($1, $2, $3, to_timestamp($4), 30382, $5::jsonb, '', $6, $7)",
+    )
+    .bind(community.as_uuid())
+    .bind(&id[..])
+    .bind(&author[..])
+    .bind(at)
+    .bind(&tags)
+    .bind(b"signature".as_slice())
+    .bind(&d)
+    .execute(pool)
+    .await
+    .expect("insert verified profile");
+    // The relay keeps the newest one per person as the head.
+    let hex = |bytes: &[u8]| bytes.iter().map(|b| format!("{b:02x}")).collect::<String>();
+    let event = format!(
+        r#"{{"id":"{}","pubkey":"{}","created_at":{at},"kind":30382,"tags":{tags},"content":"","sig":""}}"#,
+        hex(&id),
+        hex(&author)
+    );
+    sqlx::query(
+        "INSERT INTO nip_fi_profile_heads (community_id, pubkey, issuer, issued_at, event) \
+         VALUES ($1, $2, 'issuer', 0, $3::jsonb) \
+         ON CONFLICT (community_id, pubkey) DO UPDATE SET event = EXCLUDED.event",
+    )
+    .bind(community.as_uuid())
+    .bind(&d)
+    .bind(event)
+    .execute(pool)
+    .await
+    .expect("insert verified profile head");
+    id
+}
+
+/// Insert a kind:0 profile for `pubkey`; returns its event id.
+async fn put_profile_for(
+    pool: &PgPool,
+    community: CommunityId,
+    pubkey: [u8; 32],
+    content: &str,
+    at: i64,
+) -> [u8; 32] {
+    let id = rand_bytes32();
+    insert_event(pool, community, id, pubkey, 0, content, None, at).await;
+    id
+}
+
+async fn people(
+    pool: &PgPool,
+    community: CommunityId,
+    q: &str,
+    verified_by: Option<[u8; 32]>,
+    page: u32,
+    per_page: u32,
+) -> Vec<[u8; 32]> {
+    SearchService::new(pool.clone())
+        .search_profile_names(community, q, verified_by, page, per_page)
+        .await
+        .expect("profile name search")
+        .hits
+        .into_iter()
+        .map(|hit| hit.event_id)
+        .collect()
+}
+
+/// With a verifying key, people also match on the names in their current
+/// kind:30382 signed by that key. Pages count people; each person returns
+/// their kind:0 then their verified profile, whichever matched.
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn profile_names_match_verified_names() {
+    let (pool, schema) = setup().await;
+    let c = mk_community(&pool, "profile-names-verified.example").await;
+    let other = mk_community(&pool, "profile-names-verified-other.example").await;
+    let relay = rand_bytes32();
+    let forger = rand_bytes32();
+
+    let fake = rand_bytes32();
+    let fake_profile = put_profile_for(
+        &pool,
+        c,
+        fake,
+        r#"{"name":"Fake","display_name":"Fak"}"#,
+        1_700_000_000,
+    )
+    .await;
+    let fake_verified = put_verified(
+        &pool,
+        c,
+        relay,
+        fake,
+        &[
+            ("display_name", "fname"),
+            ("username", "fname"),
+            ("full_name", "Fake Name"),
+        ],
+        1_700_000_010,
+    )
+    .await;
+    // Another Fake, never verified.
+    let namesake = rand_bytes32();
+    let namesake_profile =
+        put_profile_for(&pool, c, namesake, r#"{"name":"Fake"}"#, 1_700_000_020).await;
+    // Verified, with no kind:0 at all.
+    let gina = rand_bytes32();
+    let gina_verified = put_verified(
+        &pool,
+        c,
+        relay,
+        gina,
+        &[("display_name", "gname"), ("full_name", "Name Gina")],
+        1_700_000_030,
+    )
+    .await;
+    // Signed by another key: not a verified profile.
+    let fred = rand_bytes32();
+    put_profile_for(&pool, c, fred, r#"{"name":"Fred"}"#, 1_700_000_040).await;
+    put_verified(
+        &pool,
+        c,
+        forger,
+        fred,
+        &[("full_name", "Fred Name")],
+        1_700_000_050,
+    )
+    .await;
+    // Replaced (soft-deleted) verified profile.
+    let dan = rand_bytes32();
+    put_profile_for(&pool, c, dan, r#"{"name":"Dan"}"#, 1_700_000_060).await;
+    let replaced = put_verified(
+        &pool,
+        c,
+        relay,
+        dan,
+        &[("full_name", "Dan Name")],
+        1_700_000_070,
+    )
+    .await;
+    sqlx::query("UPDATE events SET deleted_at = now() WHERE community_id = $1 AND id = $2")
+        .bind(c.as_uuid())
+        .bind(&replaced[..])
+        .execute(&pool)
+        .await
+        .expect("replace verified profile");
+    // Another community.
+    put_verified(
+        &pool,
+        other,
+        relay,
+        rand_bytes32(),
+        &[("full_name", "Olga Name")],
+        1_700_000_080,
+    )
+    .await;
+
+    // Gina is tier 1 (a name starts with "name"); Fake is tier 2.
+    let by_name = vec![gina_verified, fake_profile, fake_verified];
+    assert_eq!(people(&pool, c, "name", Some(relay), 1, 100).await, by_name);
+    assert_eq!(
+        people(&pool, c, "NAME", Some(relay), 1, 1).await,
+        by_name[..1]
+    );
+    assert_eq!(
+        people(&pool, c, "name", Some(relay), 2, 1).await,
+        by_name[1..]
+    );
+    assert!(people(&pool, c, "name", Some(relay), 3, 1).await.is_empty());
+    assert_eq!(
+        people(&pool, c, "fnam", Some(relay), 1, 100).await,
+        vec![fake_profile, fake_verified]
+    );
+    assert_eq!(
+        people(&pool, c, "name fake", Some(relay), 1, 100).await,
+        vec![fake_profile, fake_verified]
+    );
+    // Full name and username together: both are in the verified profile.
+    assert_eq!(
+        people(&pool, c, "fake fname", Some(relay), 1, 100).await,
+        vec![fake_profile, fake_verified]
+    );
+    // A kind:0 match still returns the person's verified profile.
+    let fakes = people(&pool, c, "fake", Some(relay), 1, 100).await;
+    assert_eq!(fakes.len(), 3, "{fakes:?}");
+    let at = fakes
+        .iter()
+        .position(|id| *id == fake_profile)
+        .expect("fake");
+    assert_eq!(fakes.get(at + 1), Some(&fake_verified));
+    assert!(fakes.contains(&namesake_profile));
+    // Verified names need a three-character prefix; kind:0 does not.
+    assert!(people(&pool, c, "gn", Some(relay), 1, 100).await.is_empty());
+    assert_eq!(
+        people(&pool, c, "gna", Some(relay), 1, 100).await,
+        vec![gina_verified]
+    );
+    let mut short = people(&pool, c, "fa", Some(relay), 1, 100).await;
+    short.sort();
+    let mut full = people(&pool, c, "fake", Some(relay), 1, 100).await;
+    full.sort();
+    assert_eq!(short, full);
+    // Words split between kind:0 and the verified profile do not match.
+    assert!(people(&pool, c, "fak fname", Some(relay), 1, 100)
+        .await
+        .is_empty());
+
+    // Without a verifying key, only kind:0 counts, as before.
+    assert!(people(&pool, c, "name", None, 1, 100).await.is_empty());
+    let mut fakes = people(&pool, c, "fake", None, 1, 100).await;
+    fakes.sort();
+    let mut want = vec![fake_profile, namesake_profile];
+    want.sort();
+    assert_eq!(fakes, want);
+    // The forger's key verifies only its own events.
+    assert!(people(&pool, c, "fred name", Some(relay), 1, 100)
+        .await
+        .is_empty());
+
+    // A person ranks by their single best match. Kim's exact kind:0 name
+    // outranks Ken's, and Ken's better-ranked verified match has a worse tier.
+    let ken = rand_bytes32();
+    let ken_profile = put_profile_for(&pool, c, ken, r#"{"name":"Kname"}"#, 1_700_000_100).await;
+    put_verified(
+        &pool,
+        c,
+        relay,
+        ken,
+        &[
+            ("display_name", "Doctor Kname"),
+            ("username", "doctor-kname"),
+            ("full_name", "Doctor Kname"),
+        ],
+        1_700_000_110,
+    )
+    .await;
+    let kim = rand_bytes32();
+    let kim_profile = put_profile_for(
+        &pool,
+        c,
+        kim,
+        r#"{"name":"Kname","display_name":"Kname"}"#,
+        1_700_000_000,
+    )
+    .await;
+    assert_eq!(
+        people(&pool, c, "kname", Some(relay), 1, 1).await,
+        vec![kim_profile]
+    );
+    assert_eq!(
+        people(&pool, c, "kname", Some(relay), 2, 1).await.first(),
+        Some(&ken_profile)
+    );
+
+    teardown(pool, &schema).await;
+}
+
+/// Verified-name search reads `idx_nip_fi_profile_heads_names`.
+#[tokio::test]
+#[ignore = "requires Postgres"]
+async fn profile_names_use_verified_name_index() {
+    let (setup_pool, schema) = setup().await;
+    let c = mk_community(&setup_pool, "profile-names-verified-index.example").await;
+    let relay = rand_bytes32();
+    for i in 0..2_000 {
+        put_verified(
+            &setup_pool,
+            c,
+            relay,
+            rand_bytes32(),
+            &[("full_name", &format!("Person{i} Name{i}"))],
+            1_700_000_000 + i,
+        )
+        .await;
+    }
+    sqlx::query("ANALYZE nip_fi_profile_heads")
+        .execute(&setup_pool)
+        .await
+        .expect("analyze heads");
+    // One connection, so this backend's statistics hold the search's scans.
+    let url = std::env::var("BUZZ_TEST_DATABASE_URL").unwrap_or_else(|_| TEST_DB_URL.to_string());
+    let pool = PgPoolOptions::new()
+        .max_connections(1)
+        .connect(&url)
+        .await
+        .expect("connect test database");
+    let scans = || async {
+        sqlx::query("SELECT pg_stat_force_next_flush()")
+            .execute(&pool)
+            .await
+            .expect("flush statistics");
+        sqlx::query("SELECT pg_stat_clear_snapshot()")
+            .execute(&pool)
+            .await
+            .expect("clear statistics snapshot");
+        sqlx::query_scalar::<_, i64>(
+            "SELECT idx_scan FROM pg_stat_user_indexes \
+             WHERE indexrelname = 'idx_nip_fi_profile_heads_names'",
+        )
+        .fetch_one(&pool)
+        .await
+        .expect("index statistics")
+    };
+    let before = scans().await;
+    let found = SearchService::new(pool.clone())
+        .search_profile_names(c, "name1234", Some(relay), 1, 10)
+        .await
+        .expect("profile name search")
+        .hits;
+    assert_eq!(found.len(), 1);
+    assert!(
+        scans().await > before,
+        "verified-name search did not use its index"
+    );
+    pool.close().await;
+    teardown(setup_pool, &schema).await;
+}
+
 /// Connect the way the relay does: [`buzz_search::pool_options`].
 async fn search_pool() -> PgPool {
     let url = std::env::var("BUZZ_TEST_DATABASE_URL").unwrap_or_else(|_| TEST_DB_URL.to_string());
@@ -2080,12 +2411,14 @@ async fn search_pool() -> PgPool {
 
 /// Backends in this test's database, other than the caller's, that wait on
 /// a lock inside a search. Scoped to the database so concurrent tests in
-/// sibling databases cannot change the count.
+/// sibling databases cannot change the count. Matches each search by its
+/// leading text, since `pg_stat_activity` keeps only the first 1 kB of a query.
 async fn waiting_searches(pool: &PgPool) -> i64 {
     sqlx::query_scalar(
         "SELECT count(*) FROM pg_stat_activity WHERE pid <> pg_backend_pid() \
-         AND datname = current_database() \
-         AND wait_event_type = 'Lock' AND query LIKE '%ts_rank_cd(%'",
+         AND datname = current_database() AND wait_event_type = 'Lock' \
+         AND (query LIKE 'SELECT id, kind, pubkey, channel_id, %ts_rank_cd(%' \
+           OR query LIKE 'WITH search_query AS (SELECT %')",
     )
     .fetch_one(pool)
     .await
@@ -2187,7 +2520,7 @@ async fn abandoned_search_cancels_its_statement() {
     };
     abandon_under_lock(&pool, "search", || svc.search(&message_query)).await;
     abandon_under_lock(&pool, "search_profile_names", || {
-        svc.search_profile_names(c, "proj", 1, 10)
+        svc.search_profile_names(c, "proj", None, 1, 10)
     })
     .await;
     teardown(pool, &schema).await;

docs/profile-search.md

@@ -2,8 +2,11 @@
 
 `POST /profiles/search` finds community members by profile name. It matches
 only the `name` and `display_name` string fields of each current global kind:0
-profile. It does not search `picture` (including data URLs), `about`, `nip05`,
-`banner`, or any other profile field. NIP-50 search over kind:0 matches the
+profile and, where the community advertises NIP-FI verified profiles
+(`federated_identity.public_profile.searchable` in NIP-11), the
+`display_name`, `username` and `full_name` tags of each current kind:30382
+verified profile signed by the relay key. It does not search `picture`
+(including data URLs), `about`, `nip05`, `banner`, or any other profile field. NIP-50 search over kind:0 matches the
 whole profile JSON, so it can return profiles whose avatar data or bio happens
 to contain the query text.
 
@@ -26,28 +29,56 @@ Request:
 - `page`: 1-based page of `limit` results, 1 through 1000. Defaults to 1.
 - Unknown fields are rejected with `400`.
 
-Response: a JSON array of signed kind:0 events for that page, best match
-first. A page shorter than `limit` is the last page.
+Response: a JSON array of signed events for one page of people, best match
+first. Each person contributes their current kind:0, when they have one, then
+their current relay-signed kind:30382, when verified profiles are advertised
+and they have one, whichever of the two matched. A page with fewer than
+`limit` people is the last page. Clients that read only kind:0 keep working;
+they also find people who match only by verified name.
 
 Matching is the NIP-50 prefix rule, case-insensitive. The query is split on
 whitespace. Every word except the last must equal a word of `name` or
 `display_name`; the last word may be a word prefix. Words may appear in any
 order and may be split across the two fields, so "johnson wes" finds
-"Wes Johnson". Accents are not folded.
+"Wes Johnson". A verified profile matches the same way over its three name
+tags, except that its last word must be at least three characters, as in
+NIP-50. The two sources match separately: a query whose words are split
+between a person's kind:0 and their verified profile does not match.
+Accents are not folded.
 
-Results are ordered:
+People are ordered by their best match across both sources:
 
-1. `name` or `display_name` equal to the query;
+1. a name equal to the query;
 2. a name that starts with the query;
 3. a name with a later word that starts with the query;
 4. any other match.
 
-For these tiers the query and both names are lower-cased, with whitespace
+For these tiers the query and the names are lower-cased, with whitespace
 runs collapsed to one space and trimmed, by the same database expression.
-Ties go by relevance of the query to the two names alone (not the whole
-profile, as in NIP-50), then newest first, then event id, so pages do not
+Ties go by relevance of the query to the names alone (not the whole
+profile, as in NIP-50), then newest first, then pubkey, so pages do not
 overlap. Profile content that is not a JSON object, or a name field that
 is not a string, never matches. Content that PostgreSQL cannot read as `jsonb`
 (for example a `\u0000` escape or very deep nesting) has no names, so it
 never matches and never makes the search fail
 (`schema/functions/profile_name_fields.sql`).
+
+## Deployment
+
+Verified-name matching reads `idx_nip_fi_profile_heads_names`. Migration
+`0081` builds it only while `nip_fi_profile_heads` is empty; on a populated
+database it fails until the index exists. Build it first, outside a
+transaction:
+
+```sql
+CREATE INDEX CONCURRENTLY idx_nip_fi_profile_heads_names
+    ON public.nip_fi_profile_heads USING GIN (
+    to_tsvector('simple',
+        COALESCE(jsonb_path_query_first(event, 'strict $.tags[*] ? (@[0] == "display_name")[1]', '{}', true) #>> '{}', '')
+        || ' ' || COALESCE(jsonb_path_query_first(event, 'strict $.tags[*] ? (@[0] == "username")[1]', '{}', true) #>> '{}', '')
+        || ' ' || COALESCE(jsonb_path_query_first(event, 'strict $.tags[*] ? (@[0] == "full_name")[1]', '{}', true) #>> '{}', ''))
+);
+```
+
+If the build fails, drop the invalid index and build it again. The migration
+then checks that the index is valid and has this definition.

migrations/0081_nip_fi_profile_head_names.sql

@@ -0,0 +1,27 @@
+-- Profile name search matches verified profiles by name. Index the names in
+-- each current head (crates/buzz-search/src/query.rs, verified_names_tsv).
+-- Populated databases prebuild this index concurrently before this migration.
+-- See docs/profile-search.md.
+SET LOCAL lock_timeout = '1s';
+SET LOCAL statement_timeout = '5s';
+
+DO $$
+DECLARE
+    expression CONSTANT TEXT := $expr$to_tsvector('simple'::regconfig, ((((COALESCE((jsonb_path_query_first(event, 'strict $."tags"[*]?(@[0] == "display_name")[1]'::jsonpath, '{}'::jsonb, true) #>> '{}'::text[]), ''::text) || ' '::text) || COALESCE((jsonb_path_query_first(event, 'strict $."tags"[*]?(@[0] == "username")[1]'::jsonpath, '{}'::jsonb, true) #>> '{}'::text[]), ''::text)) || ' '::text) || COALESCE((jsonb_path_query_first(event, 'strict $."tags"[*]?(@[0] == "full_name")[1]'::jsonpath, '{}'::jsonb, true) #>> '{}'::text[]), ''::text)))$expr$;
+BEGIN
+    IF to_regclass('public.idx_nip_fi_profile_heads_names') IS NULL THEN
+        IF EXISTS (SELECT 1 FROM public.nip_fi_profile_heads LIMIT 1) THEN
+            RAISE EXCEPTION 'prebuild idx_nip_fi_profile_heads_names concurrently; see docs/profile-search.md';
+        END IF;
+        EXECUTE format('CREATE INDEX idx_nip_fi_profile_heads_names ON public.nip_fi_profile_heads USING gin (%s)', expression);
+    END IF;
+    IF NOT EXISTS (
+        SELECT 1 FROM pg_index WHERE indexrelid = to_regclass('public.idx_nip_fi_profile_heads_names')
+            AND indisvalid AND indisready AND indislive
+            AND pg_get_indexdef(indexrelid) = format(
+                'CREATE INDEX idx_nip_fi_profile_heads_names ON public.nip_fi_profile_heads USING gin (%s)',
+                expression)
+    ) THEN
+        RAISE EXCEPTION 'idx_nip_fi_profile_heads_names is invalid or has a different definition';
+    END IF;
+END $$;

schema/tables/public/nip_fi_profile_heads.sql

@@ -6,3 +6,12 @@ CREATE TABLE nip_fi_profile_heads (
     event JSONB NOT NULL,
     PRIMARY KEY (community_id, pubkey)
 );
+
+-- Profile name search over verified names
+-- (crates/buzz-search/src/query.rs, verified_names_tsv).
+CREATE INDEX idx_nip_fi_profile_heads_names ON nip_fi_profile_heads USING GIN (
+    to_tsvector('simple',
+        COALESCE(jsonb_path_query_first(event, 'strict $.tags[*] ? (@[0] == "display_name")[1]', '{}', true) #>> '{}', '')
+        || ' ' || COALESCE(jsonb_path_query_first(event, 'strict $.tags[*] ? (@[0] == "username")[1]', '{}', true) #>> '{}', '')
+        || ' ' || COALESCE(jsonb_path_query_first(event, 'strict $.tags[*] ? (@[0] == "full_name")[1]', '{}', true) #>> '{}', ''))
+);