mirror of
https://github.com/block/buzz.git
synced 2026-11-13 12:51:34 +00:00
279a31daac
## What
This is phase 1 of 2 for retiring the plpgsql function
`profile_name_fields(text)`. Its only caller, profile name search
(`search_profile_names`), now gets `name` and `display_name` from an
inline SQL expression (`PROFILE_NAME_FIELDS` in
`crates/buzz-search/src/query.rs`) instead.
The function's exception handler made unreadable kind:0 content have no
names rather than fail the search. The inline version keeps that
guarantee in two steps:
- **`pg_input_is_valid(content, 'jsonb')`** rejects the soft parse
errors: a `\^@` escape, a lone surrogate, a huge number, and malformed
JSON.
- **A bracket bound for deep nesting.** PostgreSQL's JSON parser is
recursive. Past about 10k levels at the default `max_stack_depth`, it
raises `stack depth limit exceeded`, and `pg_input_is_valid` does not
catch that. I confirmed this on PG 17: depth 10k is fine and 20k raises.
So content with more than 4096 `[` and `{` is never parsed and has no
names. The count bounds the depth. Ingest already rejects kind:0 content
nested deeper than 128 levels (serde_json), so this only matters for
content stored another way.
`OFFSET 0` keeps the parse from being copied into each reference to
`doc`, so each candidate row is parsed once, plus one validity pass. The
plpgsql function also parsed each row once.
**Behavior change (edge only):** a kind:0 with more than 4096 brackets,
even inside strings, is no longer findable by name. Real profiles don't
come close, and a `picture` data URL is base64 with no brackets.
`pg_input_is_valid` needs PostgreSQL 16 or later. CI runs 16, and the
dev compose runs 17.
**Phase 2 (not here):** a migration and desired-state change drops the
function, its `schema/functions/` file and its `schema.sql` entry once
this release has shipped.
## Tests
- `profile_names_survive_unparsable_content` already stores a `\^@`
profile, a lone-surrogate profile, a `1e1000000` profile and a 60k-deep
profile next to a plain one, and requires the search to return only the
plain one. It now also covers the bound itself: a profile with exactly
4096 brackets, all but one inside a string, is still found.
- Must-fail mutants, each run on Blox: bracket bound removed (the search
fails with a stack overflow), validity check removed, bound tightened by
one, and the string-type check loosened.
## What got simpler
Profile search no longer depends on a database function. Name extraction
lives next to the rest of the search SQL it feeds.
## Verification
Blox, at `be0db303ea90033cc2c37ac8f8f07c27f4b08e31` (clean tree):
| Gate | Result |
|---|---|
| `cargo fmt --all --check` | pass |
| `cargo clippy -p buzz-search --all-targets -- -D warnings` | pass |
| `cargo nextest run -p buzz-search --lib` | 4/4 |
| `scripts/postgres-test-run.sh -p buzz-search --tests -E
'test(/profile_names/)'` | 4/4 |
| Must-fail mutants (no bound, no validity check, bound tightened by
one, non-string names allowed) | 4/4 killed |
Probe on PostgreSQL 17.11: `pg_input_is_valid` returns false for `\^@`,
a lone surrogate, `1e1000000` and truncated JSON; nesting depth 10k is
valid; depth 20k raises `stack depth limit exceeded`. CI runs the full
Postgres lanes.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
buzz-review-completed
---------
Signed-off-by: Luke Tornquist <tornquist@squareup.com>
Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2618 lines
77 KiB
Rust
2618 lines
77 KiB
Rust
//! Integration tests for community-scoped Postgres FTS.
|
||
//!
|
||
//! Run through `scripts/postgres-test-run.sh -p buzz-search --tests` for a
|
||
//! fresh desired-state database per test. Outside that lane, point
|
||
//! BUZZ_TEST_DATABASE_URL at a disposable dedicated database; fixtures use
|
||
//! the production migrator and unique community IDs.
|
||
|
||
use buzz_core::{
|
||
kind::{
|
||
AUTHOR_ONLY_KINDS, KIND_AGENT_TURN_METRIC, KIND_MEMBER_ADDED_NOTIFICATION,
|
||
KIND_MEMBER_REMOVED_NOTIFICATION, P_GATED_KINDS,
|
||
},
|
||
CommunityId,
|
||
};
|
||
use buzz_search::{ChannelScope, SearchQuery, SearchService};
|
||
use sqlx::{postgres::PgPoolOptions, PgPool};
|
||
use uuid::Uuid;
|
||
|
||
const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1 -- local test-only credentials
|
||
async fn setup() -> (PgPool, String) {
|
||
let url = std::env::var("BUZZ_TEST_DATABASE_URL").unwrap_or_else(|_| TEST_DB_URL.to_string());
|
||
// The PostgreSQL lane supplies an isolated desired-state database. Outside
|
||
// that lane use a dedicated database with the production migrations.
|
||
let pool = PgPoolOptions::new()
|
||
.max_connections(5)
|
||
.connect(&url)
|
||
.await
|
||
.expect("connect test database");
|
||
if std::env::var("BUZZ_TEST_SCHEMA_MODE").as_deref() != Ok("desired") {
|
||
buzz_db::migration::run_migrations(&pool)
|
||
.await
|
||
.expect("apply production migrations");
|
||
}
|
||
(pool, String::new())
|
||
}
|
||
|
||
async fn teardown(pool: PgPool, _schema: &str) {
|
||
pool.close().await;
|
||
// The test runner owns database cleanup.
|
||
}
|
||
|
||
/// Insert a community row, return its id.
|
||
async fn mk_community(pool: &PgPool, host: &str) -> CommunityId {
|
||
let id = Uuid::new_v4();
|
||
sqlx::query("INSERT INTO communities (id, host, signing_key) VALUES ($1, $2, $3)")
|
||
.bind(id)
|
||
.bind(host)
|
||
.bind(b"signingkey".as_slice())
|
||
.execute(pool)
|
||
.await
|
||
.expect("insert community");
|
||
CommunityId::from_uuid(id)
|
||
}
|
||
|
||
/// Insert a minimal event. `created_at_secs` is unix seconds.
|
||
#[allow(clippy::too_many_arguments)]
|
||
async fn insert_event(
|
||
pool: &PgPool,
|
||
community: CommunityId,
|
||
id: [u8; 32],
|
||
pubkey: [u8; 32],
|
||
kind: i32,
|
||
content: &str,
|
||
channel_id: Option<Uuid>,
|
||
created_at_secs: i64,
|
||
) {
|
||
sqlx::query(
|
||
"INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, channel_id) \
|
||
VALUES ($1, $2, $3, to_timestamp($4), $5, '[]'::jsonb, $6, $7, $8)",
|
||
)
|
||
.bind(community.as_uuid())
|
||
.bind(&id[..])
|
||
.bind(&pubkey[..])
|
||
.bind(created_at_secs)
|
||
.bind(kind)
|
||
.bind(content)
|
||
.bind(b"signature".as_slice())
|
||
.bind(channel_id)
|
||
.execute(pool)
|
||
.await
|
||
.expect("insert event");
|
||
}
|
||
|
||
fn rand_bytes32() -> [u8; 32] {
|
||
let mut out = [0u8; 32];
|
||
let u = Uuid::new_v4();
|
||
let bytes = u.as_bytes();
|
||
out[..16].copy_from_slice(bytes);
|
||
out[16..].copy_from_slice(bytes);
|
||
out
|
||
}
|
||
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn search_finds_event_in_same_community() {
|
||
let (pool, schema) = setup().await;
|
||
|
||
let c_a = mk_community(&pool, "a.example").await;
|
||
let evt_id = rand_bytes32();
|
||
let pk = rand_bytes32();
|
||
insert_event(
|
||
&pool,
|
||
c_a,
|
||
evt_id,
|
||
pk,
|
||
9,
|
||
"hello wonderland — buzz everyone",
|
||
None,
|
||
1700000000,
|
||
)
|
||
.await;
|
||
|
||
let svc = SearchService::new(pool.clone());
|
||
let result = svc
|
||
.search(&SearchQuery {
|
||
community: c_a,
|
||
q: "wonderland".into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: None,
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::FullText,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.expect("search ok");
|
||
|
||
assert_eq!(result.hits.len(), 1);
|
||
assert_eq!(result.hits[0].event_id, evt_id);
|
||
assert_eq!(result.hits[0].kind, 9);
|
||
assert_eq!(result.hits[0].created_at, 1700000000);
|
||
assert!(result.hits[0].rank > 0.0);
|
||
|
||
teardown(pool, &schema).await;
|
||
}
|
||
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn search_does_not_return_other_community_events() {
|
||
// The load-bearing test: event indexed under community A, query bound to
|
||
// community B → zero hits.
|
||
let (pool, schema) = setup().await;
|
||
|
||
let c_a = mk_community(&pool, "a.example").await;
|
||
let c_b = mk_community(&pool, "b.example").await;
|
||
let pk = rand_bytes32();
|
||
insert_event(
|
||
&pool,
|
||
c_a,
|
||
rand_bytes32(),
|
||
pk,
|
||
9,
|
||
"only-in-a unique-token-xyz",
|
||
None,
|
||
1700000000,
|
||
)
|
||
.await;
|
||
|
||
let svc = SearchService::new(pool.clone());
|
||
let result_a = svc
|
||
.search(&SearchQuery {
|
||
community: c_a,
|
||
q: "unique-token-xyz".into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: None,
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::FullText,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.unwrap();
|
||
assert_eq!(result_a.hits.len(), 1, "A should see its own event");
|
||
|
||
let result_b = svc
|
||
.search(&SearchQuery {
|
||
community: c_b,
|
||
q: "unique-token-xyz".into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: None,
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::FullText,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.unwrap();
|
||
assert_eq!(result_b.hits.len(), 0, "B must not see A's event");
|
||
|
||
teardown(pool, &schema).await;
|
||
}
|
||
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn kind0_search_by_display_name_works_without_flattening() {
|
||
// The case I worried might regress without the kind:0 content-flattening
|
||
// hack. Postgres FTS over raw JSON content tokenizes through the
|
||
// punctuation and finds display_name/nip05 values.
|
||
let (pool, schema) = setup().await;
|
||
|
||
let c = mk_community(&pool, "a.example").await;
|
||
let pk = rand_bytes32();
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
pk,
|
||
0,
|
||
r#"{"display_name":"Alice Wonderland","name":"alice","nip05":"alice@buzz.app","about":"hello"}"#,
|
||
None,
|
||
1700000000,
|
||
)
|
||
.await;
|
||
|
||
let svc = SearchService::new(pool.clone());
|
||
for q in ["wonderland", "alice", "alice@buzz.app"] {
|
||
let r = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: q.to_string(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: Some(vec![0]),
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::FullText,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.unwrap();
|
||
assert_eq!(r.hits.len(), 1, "kind:0 query {q:?} should find Alice");
|
||
}
|
||
|
||
teardown(pool, &schema).await;
|
||
}
|
||
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn short_kind0_prefix_prioritizes_exact_lexeme_on_a_noisy_page() {
|
||
let (pool, schema) = setup().await;
|
||
|
||
let c = mk_community(&pool, "short-profile-prefix.example").await;
|
||
let exact_id = rand_bytes32();
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
exact_id,
|
||
rand_bytes32(),
|
||
0,
|
||
r#"{"display_name":"jm"}"#,
|
||
None,
|
||
1_700_000_000,
|
||
)
|
||
.await;
|
||
|
||
// These profiles all match jm:* and are newer than the exact name. Without
|
||
// exact-lexeme priority they consume the entire bounded first page.
|
||
for (i, display_name) in ["jma", "jmbravo", "jmcharlie", "jmdelta"]
|
||
.iter()
|
||
.enumerate()
|
||
{
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
rand_bytes32(),
|
||
0,
|
||
&format!(r#"{{"display_name":"{display_name}"}}"#),
|
||
None,
|
||
1_700_000_100 + i as i64,
|
||
)
|
||
.await;
|
||
}
|
||
|
||
let svc = SearchService::new(pool.clone());
|
||
let first_page = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: "jm".into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: Some(vec![0]),
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 3,
|
||
mode: buzz_search::SearchMode::Prefix,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.expect("short profile prefix search ok");
|
||
|
||
assert_eq!(first_page.hits.len(), 3);
|
||
assert_eq!(first_page.hits[0].event_id, exact_id);
|
||
|
||
teardown(pool, &schema).await;
|
||
}
|
||
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn prefix_mode_matches_final_token_prefix_without_changing_full_text() {
|
||
let (pool, schema) = setup().await;
|
||
|
||
let c = mk_community(&pool, "prefix.example").await;
|
||
let project_plan = rand_bytes32();
|
||
let project_archive = rand_bytes32();
|
||
let projectile_plan = rand_bytes32();
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
project_plan,
|
||
rand_bytes32(),
|
||
9,
|
||
"project planning milestone",
|
||
None,
|
||
1_700_000_000,
|
||
)
|
||
.await;
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
project_archive,
|
||
rand_bytes32(),
|
||
9,
|
||
"project archive",
|
||
None,
|
||
1_700_000_001,
|
||
)
|
||
.await;
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
projectile_plan,
|
||
rand_bytes32(),
|
||
9,
|
||
"projectile planning distractor",
|
||
None,
|
||
1_700_000_002,
|
||
)
|
||
.await;
|
||
// Rows whose own parser output contains a short lexeme, so a short
|
||
// trailing `:*` from punctuation in the query would match them.
|
||
let contraction = rand_bytes32();
|
||
let hyphenated = rand_bytes32();
|
||
let profile = rand_bytes32();
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
contraction,
|
||
rand_bytes32(),
|
||
9,
|
||
"don't panic",
|
||
None,
|
||
1_700_000_003,
|
||
)
|
||
.await;
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
hyphenated,
|
||
rand_bytes32(),
|
||
9,
|
||
"pre-processing done",
|
||
None,
|
||
1_700_000_004,
|
||
)
|
||
.await;
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
profile,
|
||
rand_bytes32(),
|
||
0,
|
||
r#"{"display_name":"prof"}"#,
|
||
None,
|
||
1_700_000_005,
|
||
)
|
||
.await;
|
||
|
||
let svc = SearchService::new(pool.clone());
|
||
let full_text = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: "pro".into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: Some(vec![9]),
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::FullText,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.expect("full text search ok");
|
||
assert!(
|
||
full_text.hits.is_empty(),
|
||
"plain full-text search must stay word/lexeme-based, got {:?}",
|
||
full_text
|
||
.hits
|
||
.iter()
|
||
.map(|h| h.event_id)
|
||
.collect::<Vec<_>>()
|
||
);
|
||
|
||
let prefix = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: "pro".into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: Some(vec![9]),
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::Prefix,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.expect("prefix search ok");
|
||
let prefix_ids: Vec<[u8; 32]> = prefix.hits.iter().map(|h| h.event_id).collect();
|
||
assert!(prefix_ids.contains(&project_plan));
|
||
assert!(prefix_ids.contains(&project_archive));
|
||
|
||
// Non-profile prefix search returns no hits, in either sort, whatever the
|
||
// kinds and whatever completed words precede it, when the trailing token's
|
||
// `simple` parser output has a lexeme below three characters. Punctuation
|
||
// must not pad the count: `pr.`, `@pr` and `#pr` still yield `pr`, `don't`
|
||
// yields `t` and `pre-p` yields `p`. Each row would match without the
|
||
// minimum. kind:0-only stays one-character typeahead.
|
||
for (q, kinds, sort, expected) in [
|
||
(
|
||
"pr",
|
||
Some(vec![9]),
|
||
buzz_search::SearchSort::Relevance,
|
||
vec![],
|
||
),
|
||
(
|
||
" p ",
|
||
Some(vec![9]),
|
||
buzz_search::SearchSort::Relevance,
|
||
vec![],
|
||
),
|
||
(
|
||
"project pr",
|
||
Some(vec![9]),
|
||
buzz_search::SearchSort::Relevance,
|
||
vec![],
|
||
),
|
||
(
|
||
"project p ",
|
||
Some(vec![9]),
|
||
buzz_search::SearchSort::Relevance,
|
||
vec![],
|
||
),
|
||
("pr", None, buzz_search::SearchSort::Relevance, vec![]),
|
||
(
|
||
"pr",
|
||
Some(vec![0, 9]),
|
||
buzz_search::SearchSort::Relevance,
|
||
vec![],
|
||
),
|
||
("pr", Some(vec![9]), buzz_search::SearchSort::Newest, vec![]),
|
||
(
|
||
"pr.",
|
||
Some(vec![9]),
|
||
buzz_search::SearchSort::Relevance,
|
||
vec![],
|
||
),
|
||
(
|
||
"@pr",
|
||
Some(vec![9]),
|
||
buzz_search::SearchSort::Relevance,
|
||
vec![],
|
||
),
|
||
(
|
||
"#pr",
|
||
Some(vec![9]),
|
||
buzz_search::SearchSort::Relevance,
|
||
vec![],
|
||
),
|
||
(
|
||
"...p",
|
||
Some(vec![9]),
|
||
buzz_search::SearchSort::Relevance,
|
||
vec![],
|
||
),
|
||
(
|
||
"don't",
|
||
Some(vec![9]),
|
||
buzz_search::SearchSort::Relevance,
|
||
vec![],
|
||
),
|
||
(
|
||
"pre-p",
|
||
Some(vec![9]),
|
||
buzz_search::SearchSort::Relevance,
|
||
vec![],
|
||
),
|
||
// Controls: padded but long enough, punctuation-only trailing token
|
||
// (no `:*` lexeme, so the completed word matches exactly), long
|
||
// hyphenated parts, a short completed word (exact, so not limited),
|
||
// and a one-character profile query.
|
||
(
|
||
"pro.",
|
||
Some(vec![9]),
|
||
buzz_search::SearchSort::Relevance,
|
||
vec![project_plan, project_archive, projectile_plan, hyphenated],
|
||
),
|
||
(
|
||
"project ?",
|
||
Some(vec![9]),
|
||
buzz_search::SearchSort::Relevance,
|
||
vec![project_plan, project_archive],
|
||
),
|
||
(
|
||
"pre-pro",
|
||
Some(vec![9]),
|
||
buzz_search::SearchSort::Relevance,
|
||
vec![hyphenated],
|
||
),
|
||
(
|
||
"t pan",
|
||
Some(vec![9]),
|
||
buzz_search::SearchSort::Relevance,
|
||
vec![contraction],
|
||
),
|
||
(
|
||
"p",
|
||
Some(vec![0]),
|
||
buzz_search::SearchSort::Relevance,
|
||
vec![profile],
|
||
),
|
||
] {
|
||
let result = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: q.into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: kinds.clone(),
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::Prefix,
|
||
sort,
|
||
})
|
||
.await
|
||
.expect("prefix search ok");
|
||
let mut got: Vec<[u8; 32]> = result.hits.iter().map(|h| h.event_id).collect();
|
||
let mut want = expected.clone();
|
||
got.sort();
|
||
want.sort();
|
||
assert_eq!(got, want, "{q:?} {kinds:?} {sort:?}");
|
||
}
|
||
// A short trailing lexeme is a prefix-mode rule only; full text still
|
||
// matches the contraction exactly.
|
||
let full_text_contraction = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: "don't".into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: Some(vec![9]),
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::FullText,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.expect("full text search ok");
|
||
assert_eq!(
|
||
full_text_contraction
|
||
.hits
|
||
.iter()
|
||
.map(|h| h.event_id)
|
||
.collect::<Vec<_>>(),
|
||
vec![contraction]
|
||
);
|
||
|
||
let multi_token_prefix = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: "project pla".into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: Some(vec![9]),
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::Prefix,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.expect("multi-token prefix search ok");
|
||
assert_eq!(multi_token_prefix.hits.len(), 1);
|
||
assert_eq!(multi_token_prefix.hits[0].event_id, project_plan);
|
||
|
||
let completed_token_is_exact = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: "project plan".into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: Some(vec![9]),
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::Prefix,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.expect("completed-token exact + trailing-prefix search ok");
|
||
let completed_ids: Vec<[u8; 32]> = completed_token_is_exact
|
||
.hits
|
||
.iter()
|
||
.map(|h| h.event_id)
|
||
.collect();
|
||
assert!(completed_ids.contains(&project_plan));
|
||
assert!(
|
||
!completed_ids.contains(&projectile_plan),
|
||
"completed tokens must stay exact; only the trailing token gets :*"
|
||
);
|
||
|
||
teardown(pool, &schema).await;
|
||
}
|
||
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn prefix_mode_handles_tsquery_boundary_punctuation() {
|
||
let (pool, schema) = setup().await;
|
||
|
||
let c = mk_community(&pool, "prefix-punctuation.example").await;
|
||
let hit_id = rand_bytes32();
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
hit_id,
|
||
rand_bytes32(),
|
||
9,
|
||
"operators ' : & | ( ) ! alpha beta marker",
|
||
None,
|
||
1_700_000_000,
|
||
)
|
||
.await;
|
||
|
||
let svc = SearchService::new(pool.clone());
|
||
let result = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: "operators ' : & | ( ) ! alpha bet".into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: Some(vec![9]),
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::Prefix,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.expect("prefix punctuation search ok");
|
||
|
||
assert_eq!(result.hits.len(), 1);
|
||
assert_eq!(result.hits[0].event_id, hit_id);
|
||
|
||
teardown(pool, &schema).await;
|
||
}
|
||
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn prefix_mode_preserves_storage_level_privacy_exclusions() {
|
||
let (pool, schema) = setup().await;
|
||
|
||
let c = mk_community(&pool, "prefix-privacy.example").await;
|
||
let token = "prefixprivacy_unique_marker";
|
||
let control_id = rand_bytes32();
|
||
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
control_id,
|
||
rand_bytes32(),
|
||
9,
|
||
&format!("public control {token}"),
|
||
None,
|
||
1_700_000_000,
|
||
)
|
||
.await;
|
||
|
||
for (i, &kind) in [1059_u32, 30300, 30622, 44100, 44101].iter().enumerate() {
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
rand_bytes32(),
|
||
kind as i32,
|
||
&format!("private kind {kind} {token}"),
|
||
None,
|
||
1_700_000_100 + i as i64,
|
||
)
|
||
.await;
|
||
}
|
||
|
||
let svc = SearchService::new(pool.clone());
|
||
let result = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: "prefixpriv".into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: None,
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::Prefix,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.expect("prefix privacy search ok");
|
||
|
||
assert_eq!(result.hits.len(), 1);
|
||
assert_eq!(result.hits[0].event_id, control_id);
|
||
|
||
teardown(pool, &schema).await;
|
||
}
|
||
|
||
/// Prefix relevance (typeahead) ranks only the newest 1000 matches, so an
|
||
/// older, better-ranked message drops out, and its pages end with that
|
||
/// window; full-text and profile relevance still rank every match.
|
||
/// Mutation: rank every prefix match → the old message tops the first page;
|
||
/// window profiles too → the old profile drops off the first page.
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn prefix_relevance_ranks_only_the_newest_matches() {
|
||
let (pool, schema) = setup().await;
|
||
let c = mk_community(&pool, "rank-window.example").await;
|
||
let (old, old_profile) = (rand_bytes32(), rand_bytes32());
|
||
for (id, kind) in [(old, 9), (old_profile, 0)] {
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
id,
|
||
rand_bytes32(),
|
||
kind,
|
||
"planning planning planning planning",
|
||
None,
|
||
1_700_000_000,
|
||
)
|
||
.await;
|
||
// 1000 newer, weaker matches fill the window.
|
||
sqlx::query(
|
||
"INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig) \
|
||
SELECT $1, sha256((g || '/' || $2)::bytea), sha256((g || '/' || $2)::bytea), \
|
||
to_timestamp(1700000001 + g), $2, '[]'::jsonb, 'planning notes ' || g, 'signature' \
|
||
FROM generate_series(1, 1000) g",
|
||
)
|
||
.bind(c.as_uuid())
|
||
.bind(kind)
|
||
.execute(&pool)
|
||
.await
|
||
.expect("insert newer matches");
|
||
}
|
||
|
||
let svc = SearchService::new(pool.clone());
|
||
let query = |q: &str, mode, page| SearchQuery {
|
||
community: c,
|
||
q: q.into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: Some(vec![9]),
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page,
|
||
per_page: 20,
|
||
mode,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
};
|
||
let ids = |r: buzz_search::SearchResult| r.hits.iter().map(|h| h.event_id).collect::<Vec<_>>();
|
||
|
||
let typeahead = ids(svc
|
||
.search(&query("plan", buzz_search::SearchMode::Prefix, 1))
|
||
.await
|
||
.expect("prefix"));
|
||
assert_eq!(typeahead.len(), 20);
|
||
assert!(!typeahead.contains(&old), "outside the newest-1000 window");
|
||
|
||
let full_text = ids(svc
|
||
.search(&query("planning", buzz_search::SearchMode::FullText, 1))
|
||
.await
|
||
.expect("full text"));
|
||
assert_eq!(full_text.first(), Some(&old), "full text ranks every match");
|
||
|
||
// Every page ranks the same window, so prefix pages end with it.
|
||
let last = ids(svc
|
||
.search(&query("plan", buzz_search::SearchMode::Prefix, 50))
|
||
.await
|
||
.expect("page 50"));
|
||
assert_eq!(last.len(), 20);
|
||
assert!(!last.contains(&old));
|
||
let past = svc
|
||
.search(&query("plan", buzz_search::SearchMode::Prefix, 51))
|
||
.await
|
||
.expect("page 51");
|
||
assert!(past.hits.is_empty(), "no page past the window");
|
||
|
||
let profiles = ids(svc
|
||
.search(&SearchQuery {
|
||
kinds: Some(vec![0]),
|
||
..query("plan", buzz_search::SearchMode::Prefix, 1)
|
||
})
|
||
.await
|
||
.expect("profile prefix"));
|
||
assert_eq!(
|
||
profiles.first(),
|
||
Some(&old_profile),
|
||
"profile typeahead ranks every match"
|
||
);
|
||
|
||
teardown(pool, &schema).await;
|
||
}
|
||
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn channel_scope_restricts_results() {
|
||
let (pool, schema) = setup().await;
|
||
|
||
let c = mk_community(&pool, "a.example").await;
|
||
// Channels need community_id since channels PK is (community_id, id).
|
||
let ch_a = Uuid::new_v4();
|
||
let ch_b = Uuid::new_v4();
|
||
sqlx::query("INSERT INTO channels (community_id, id, name, channel_type, created_by) VALUES ($1, $2, $3, 'stream'::channel_type, $4), ($1, $5, $6, 'stream'::channel_type, $4)")
|
||
.bind(c.as_uuid())
|
||
.bind(ch_a)
|
||
.bind("ch-a")
|
||
.bind(b"\x01".repeat(32))
|
||
.bind(ch_b)
|
||
.bind("ch-b")
|
||
.execute(&pool)
|
||
.await
|
||
.expect("insert channels");
|
||
|
||
let pk = rand_bytes32();
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
pk,
|
||
9,
|
||
"shared-token in ch-a",
|
||
Some(ch_a),
|
||
1700000000,
|
||
)
|
||
.await;
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
pk,
|
||
9,
|
||
"shared-token in ch-b",
|
||
Some(ch_b),
|
||
1700000001,
|
||
)
|
||
.await;
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
pk,
|
||
9,
|
||
"shared-token global",
|
||
None,
|
||
1700000002,
|
||
)
|
||
.await;
|
||
|
||
let svc = SearchService::new(pool.clone());
|
||
|
||
// restrict to ch_a, exclude global
|
||
let r = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: "shared-token".into(),
|
||
channel_scope: ChannelScope::Channels(vec![ch_a]),
|
||
kinds: None,
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::FullText,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.unwrap();
|
||
assert_eq!(r.hits.len(), 1);
|
||
assert_eq!(r.hits[0].channel_id, Some(ch_a));
|
||
|
||
// restrict to ch_a + include global
|
||
let r = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: "shared-token".into(),
|
||
channel_scope: ChannelScope::ChannelsOrChannelLess(vec![ch_a]),
|
||
kinds: None,
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::FullText,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.unwrap();
|
||
assert_eq!(r.hits.len(), 2);
|
||
|
||
// no channel constraint
|
||
let r = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: "shared-token".into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: None,
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::FullText,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.unwrap();
|
||
assert_eq!(r.hits.len(), 3);
|
||
|
||
// empty accessible channels + exclude global = zero
|
||
let r = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: "shared-token".into(),
|
||
channel_scope: ChannelScope::Channels(vec![]),
|
||
kinds: None,
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::FullText,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.unwrap();
|
||
assert_eq!(r.hits.len(), 0);
|
||
|
||
teardown(pool, &schema).await;
|
||
}
|
||
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn deleted_events_are_excluded() {
|
||
let (pool, schema) = setup().await;
|
||
|
||
let c = mk_community(&pool, "a.example").await;
|
||
let evt_id = rand_bytes32();
|
||
let pk = rand_bytes32();
|
||
insert_event(&pool, c, evt_id, pk, 9, "deleted-token-q", None, 1700000000).await;
|
||
|
||
// Soft-delete
|
||
sqlx::query("UPDATE events SET deleted_at = NOW() WHERE community_id = $1 AND id = $2")
|
||
.bind(c.as_uuid())
|
||
.bind(&evt_id[..])
|
||
.execute(&pool)
|
||
.await
|
||
.expect("delete");
|
||
|
||
let svc = SearchService::new(pool.clone());
|
||
let r = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: "deleted-token-q".into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: None,
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::FullText,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.unwrap();
|
||
assert!(
|
||
r.hits.is_empty(),
|
||
"soft-deleted events must not appear in FTS"
|
||
);
|
||
|
||
teardown(pool, &schema).await;
|
||
}
|
||
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn empty_query_returns_empty_result_no_roundtrip() {
|
||
let (pool, schema) = setup().await;
|
||
let c = mk_community(&pool, "a.example").await;
|
||
let svc = SearchService::new(pool.clone());
|
||
|
||
for q in ["", " "] {
|
||
let r = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: q.into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: None,
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::FullText,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.unwrap();
|
||
assert!(
|
||
r.hits.is_empty(),
|
||
"empty/whitespace query must return no hits"
|
||
);
|
||
}
|
||
|
||
teardown(pool, &schema).await;
|
||
}
|
||
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn since_until_filters() {
|
||
let (pool, schema) = setup().await;
|
||
|
||
let c = mk_community(&pool, "a.example").await;
|
||
let pk = rand_bytes32();
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
pk,
|
||
9,
|
||
"time-token-zz at A",
|
||
None,
|
||
1_700_000_000,
|
||
)
|
||
.await;
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
pk,
|
||
9,
|
||
"time-token-zz at B",
|
||
None,
|
||
1_700_010_000,
|
||
)
|
||
.await;
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
pk,
|
||
9,
|
||
"time-token-zz at C",
|
||
None,
|
||
1_700_020_000,
|
||
)
|
||
.await;
|
||
|
||
let svc = SearchService::new(pool.clone());
|
||
|
||
let r = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: "time-token-zz".into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: None,
|
||
authors: None,
|
||
since: Some(1_700_005_000),
|
||
until: Some(1_700_015_000),
|
||
page: 1,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::FullText,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.unwrap();
|
||
assert_eq!(r.hits.len(), 1);
|
||
assert_eq!(r.hits[0].created_at, 1_700_010_000);
|
||
|
||
teardown(pool, &schema).await;
|
||
}
|
||
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn pagination_works() {
|
||
let (pool, schema) = setup().await;
|
||
|
||
let c = mk_community(&pool, "a.example").await;
|
||
let pk = rand_bytes32();
|
||
for i in 0..7 {
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
pk,
|
||
9,
|
||
"page-token-qq",
|
||
None,
|
||
1_700_000_000 + i,
|
||
)
|
||
.await;
|
||
}
|
||
|
||
let svc = SearchService::new(pool.clone());
|
||
|
||
let r = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: "page-token-qq".into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: None,
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 3,
|
||
mode: buzz_search::SearchMode::FullText,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.unwrap();
|
||
assert_eq!(r.hits.len(), 3);
|
||
|
||
let r2 = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: "page-token-qq".into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: None,
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 3,
|
||
per_page: 3,
|
||
mode: buzz_search::SearchMode::FullText,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.unwrap();
|
||
assert_eq!(
|
||
r2.hits.len(),
|
||
1,
|
||
"page 3 of 7 with per_page=3 should have 1 hit"
|
||
);
|
||
|
||
teardown(pool, &schema).await;
|
||
}
|
||
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn newest_sort_pages_by_time_then_id_within_scope() {
|
||
let (pool, schema) = setup().await;
|
||
|
||
let c = mk_community(&pool, "a.example").await;
|
||
let channel = Uuid::new_v4();
|
||
let pk = rand_bytes32();
|
||
let most_relevant = [9u8; 32];
|
||
let tie_low = [1u8; 32];
|
||
let tie_high = [2u8; 32];
|
||
let newest = [3u8; 32];
|
||
let rows = [
|
||
(
|
||
most_relevant,
|
||
"sortword sortword sortword sortword",
|
||
1_700_000_000,
|
||
),
|
||
(tie_high, "sortword once", 1_700_000_100),
|
||
(tie_low, "sortword once", 1_700_000_100),
|
||
(newest, "sortword once", 1_700_000_200),
|
||
];
|
||
for (id, content, created_at) in rows {
|
||
insert_event(&pool, c, id, pk, 9, content, Some(channel), created_at).await;
|
||
}
|
||
// Newer than every scoped row, but outside the requested channel scope.
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
pk,
|
||
9,
|
||
"sortword once",
|
||
Some(Uuid::new_v4()),
|
||
1_700_000_300,
|
||
)
|
||
.await;
|
||
|
||
let svc = SearchService::new(pool.clone());
|
||
let page = |page, sort| {
|
||
let svc = svc.clone();
|
||
async move {
|
||
svc.search(&SearchQuery {
|
||
community: c,
|
||
q: "sortword".into(),
|
||
channel_scope: ChannelScope::Channels(vec![channel]),
|
||
kinds: None,
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page,
|
||
per_page: 2,
|
||
mode: buzz_search::SearchMode::FullText,
|
||
sort,
|
||
})
|
||
.await
|
||
.unwrap()
|
||
.hits
|
||
.into_iter()
|
||
.map(|hit| hit.event_id)
|
||
.collect::<Vec<_>>()
|
||
}
|
||
};
|
||
|
||
let newest_order = [
|
||
page(1, buzz_search::SearchSort::Newest).await,
|
||
page(2, buzz_search::SearchSort::Newest).await,
|
||
page(3, buzz_search::SearchSort::Newest).await,
|
||
]
|
||
.concat();
|
||
assert_eq!(newest_order, vec![newest, tie_low, tie_high, most_relevant]);
|
||
assert_eq!(
|
||
page(1, buzz_search::SearchSort::Relevance).await[0],
|
||
most_relevant
|
||
);
|
||
|
||
teardown(pool, &schema).await;
|
||
}
|
||
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn channel_less_only_excludes_per_channel_events() {
|
||
// Closes the row-3 fence hole: in the legacy 2x2 shape, both
|
||
// `Some(vec![]) + true` and `None + true` silently broadened to all
|
||
// community channels rather than restricting to channel-less events.
|
||
// `ChannelScope::ChannelLessOnly` is the variant that the old type
|
||
// could not express.
|
||
//
|
||
// Adversarial check: mutate this test's expectation to `>= 2` and the
|
||
// assertion goes red against the new SQL `AND channel_id IS NULL`,
|
||
// proving the predicate bites. Mutate `query.rs` `ChannelLessOnly` arm
|
||
// to a no-op (the `Any` semantic the old code emitted) and this test
|
||
// also goes red — three hits instead of one — proving the fix is the
|
||
// emitted predicate, not the variant name.
|
||
let (pool, schema) = setup().await;
|
||
|
||
let c = mk_community(&pool, "a.example").await;
|
||
let ch_a = Uuid::new_v4();
|
||
let ch_b = Uuid::new_v4();
|
||
sqlx::query("INSERT INTO channels (community_id, id, name, channel_type, created_by) VALUES ($1, $2, $3, 'stream'::channel_type, $4), ($1, $5, $6, 'stream'::channel_type, $4)")
|
||
.bind(c.as_uuid())
|
||
.bind(ch_a)
|
||
.bind("ch-a")
|
||
.bind(b"\x01".repeat(32))
|
||
.bind(ch_b)
|
||
.bind("ch-b")
|
||
.execute(&pool)
|
||
.await
|
||
.expect("insert channels");
|
||
|
||
let pk = rand_bytes32();
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
pk,
|
||
9,
|
||
"fence-token in ch-a",
|
||
Some(ch_a),
|
||
1_700_000_000,
|
||
)
|
||
.await;
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
pk,
|
||
9,
|
||
"fence-token in ch-b",
|
||
Some(ch_b),
|
||
1_700_000_001,
|
||
)
|
||
.await;
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
pk,
|
||
9,
|
||
"fence-token channel-less",
|
||
None,
|
||
1_700_000_002,
|
||
)
|
||
.await;
|
||
|
||
let svc = SearchService::new(pool.clone());
|
||
|
||
let r = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: "fence-token".into(),
|
||
channel_scope: ChannelScope::ChannelLessOnly,
|
||
kinds: None,
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::FullText,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.unwrap();
|
||
assert_eq!(
|
||
r.hits.len(),
|
||
1,
|
||
"ChannelLessOnly must return only the channel_id IS NULL row"
|
||
);
|
||
assert_eq!(r.hits[0].channel_id, None);
|
||
|
||
teardown(pool, &schema).await;
|
||
}
|
||
|
||
/// Search-input hardening: NUL bytes are not valid Postgres text-search input.
|
||
/// Sanitize before calling `websearch_to_tsquery` so the bridge does not turn
|
||
/// adversarial search text into HTTP 500s.
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn nul_bytes_in_query_are_sanitized() {
|
||
let (pool, schema) = setup().await;
|
||
|
||
let c = mk_community(&pool, "nul.example").await;
|
||
let evt_id = rand_bytes32();
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
evt_id,
|
||
rand_bytes32(),
|
||
9,
|
||
"foo bar search text",
|
||
None,
|
||
1_700_000_000,
|
||
)
|
||
.await;
|
||
|
||
let svc = SearchService::new(pool.clone());
|
||
let result = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: "foo\0bar".into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: None,
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::FullText,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.expect("NUL-containing search query should not bubble a Postgres error");
|
||
|
||
assert_eq!(result.hits.len(), 1);
|
||
assert_eq!(result.hits[0].event_id, evt_id);
|
||
|
||
teardown(pool, &schema).await;
|
||
}
|
||
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn enormous_page_number_is_clamped() {
|
||
let (pool, schema) = setup().await;
|
||
|
||
let c = mk_community(&pool, "page-clamp.example").await;
|
||
for i in 0..5 {
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
rand_bytes32(),
|
||
9,
|
||
"clamp-token",
|
||
None,
|
||
1_700_000_000 + i,
|
||
)
|
||
.await;
|
||
}
|
||
|
||
let svc = SearchService::new(pool.clone());
|
||
let result = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: "clamp-token".into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: None,
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: u32::MAX,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::FullText,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.expect("huge page number should be bounded, not error");
|
||
|
||
assert_eq!(result.page, 1000);
|
||
assert!(result.hits.is_empty());
|
||
|
||
teardown(pool, &schema).await;
|
||
}
|
||
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn very_long_query_is_bounded_before_pg_parse() {
|
||
let (pool, schema) = setup().await;
|
||
|
||
let c = mk_community(&pool, "long-query.example").await;
|
||
let svc = SearchService::new(pool.clone());
|
||
let result = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: "x".repeat(10_000),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: None,
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::FullText,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.expect("long search query should be capped before Postgres parses it");
|
||
|
||
assert!(result.hits.is_empty());
|
||
|
||
teardown(pool, &schema).await;
|
||
}
|
||
|
||
/// Privacy regression gate: the storage layer MUST NOT make these kinds
|
||
/// searchable. The migration's `search_tsv` generated column emits NULL
|
||
/// for excluded kinds, so a `search_tsv @@ query` probe never matches.
|
||
///
|
||
/// Set kept in sync with the pre-rewrite skip in `handlers/event.rs:287-290`
|
||
/// on `main`:
|
||
/// - 1059 = `KIND_GIFT_WRAP` (NIP-17 ciphertext)
|
||
/// - 30300 = `KIND_EVENT_REMINDER` (in `AUTHOR_ONLY_KINDS`)
|
||
/// - 30622 = `KIND_DM_VISIBILITY` (per-viewer private hide state)
|
||
/// - 44100 = `KIND_MEMBER_ADDED_NOTIFICATION` (p-gated membership notice)
|
||
/// - 44101 = `KIND_MEMBER_REMOVED_NOTIFICATION` (p-gated membership notice)
|
||
/// - 44200 = `KIND_AGENT_TURN_METRIC` (NIP-AM: p-gated encrypted turn metrics)
|
||
///
|
||
/// All seven events are inserted with the same unique token in their content
|
||
/// so a single search query exercises every kind in one round-trip. Only
|
||
/// the kind:9 control must surface — the excluded kinds must not.
|
||
///
|
||
/// Mutate-bite: drop the `CASE WHEN kind IN (…)` from the generated column
|
||
/// (revert to `to_tsvector('simple', content)`) → excluded events surface →
|
||
/// restore.
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn excluded_kinds_are_storage_level_unsearchable() {
|
||
let (pool, schema) = setup().await;
|
||
|
||
let c = mk_community(&pool, "privacy.example").await;
|
||
let token = "privacykinds_unique_marker_xyzzy";
|
||
|
||
// kind:9 control — MUST be searchable.
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
rand_bytes32(),
|
||
9,
|
||
&format!("public chat — {token}"),
|
||
None,
|
||
1_700_000_000,
|
||
)
|
||
.await;
|
||
|
||
// kind:1059 gift wrap (NIP-17 ciphertext) — MUST NOT be searchable.
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
rand_bytes32(),
|
||
1059,
|
||
&format!("gift wrap — {token}"),
|
||
None,
|
||
1_700_000_001,
|
||
)
|
||
.await;
|
||
|
||
// kind:30300 event reminder (AUTHOR_ONLY_KINDS) — MUST NOT be searchable.
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
rand_bytes32(),
|
||
30300,
|
||
&format!("reminder — {token}"),
|
||
None,
|
||
1_700_000_002,
|
||
)
|
||
.await;
|
||
|
||
// kind:30622 DM visibility snapshot — MUST NOT be searchable.
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
rand_bytes32(),
|
||
30622,
|
||
&format!("dm visibility — {token}"),
|
||
None,
|
||
1_700_000_003,
|
||
)
|
||
.await;
|
||
|
||
// kind:44100 member-added notification — p-gated and MUST NOT be searchable.
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
rand_bytes32(),
|
||
KIND_MEMBER_ADDED_NOTIFICATION as i32,
|
||
&format!("member added — {token}"),
|
||
None,
|
||
1_700_000_004,
|
||
)
|
||
.await;
|
||
|
||
// kind:44101 member-removed notification — p-gated and MUST NOT be searchable.
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
rand_bytes32(),
|
||
KIND_MEMBER_REMOVED_NOTIFICATION as i32,
|
||
&format!("member removed — {token}"),
|
||
None,
|
||
1_700_000_005,
|
||
)
|
||
.await;
|
||
|
||
// kind:44200 agent turn metric — p-gated NIP-44 ciphertext and MUST NOT be searchable.
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
rand_bytes32(),
|
||
KIND_AGENT_TURN_METRIC as i32,
|
||
&format!("agent turn metric — {token}"),
|
||
None,
|
||
1_700_000_006,
|
||
)
|
||
.await;
|
||
|
||
let svc = SearchService::new(pool.clone());
|
||
let result = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: token.into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: None,
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::FullText,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.expect("search ok");
|
||
|
||
let kinds: Vec<i32> = result.hits.iter().map(|h| h.kind).collect();
|
||
|
||
// Positive: kind:9 surfaces (control — proves the search index works at all).
|
||
assert!(
|
||
kinds.contains(&9),
|
||
"kind:9 control row MUST be searchable, got kinds={kinds:?}",
|
||
);
|
||
|
||
// Negative (load-bearing): each excluded kind MUST NOT surface.
|
||
for forbidden in [
|
||
1059,
|
||
30300,
|
||
30622,
|
||
KIND_MEMBER_ADDED_NOTIFICATION as i32,
|
||
KIND_MEMBER_REMOVED_NOTIFICATION as i32,
|
||
KIND_AGENT_TURN_METRIC as i32,
|
||
] {
|
||
assert!(
|
||
!kinds.contains(&forbidden),
|
||
"kind:{forbidden} MUST NOT be searchable — \
|
||
privacy regression in `search_tsv` generated column. kinds={kinds:?}",
|
||
);
|
||
}
|
||
|
||
// Tight bound: exactly one hit (the control). Catches any future
|
||
// weakening where some-but-not-all excluded kinds surface.
|
||
assert_eq!(
|
||
result.hits.len(),
|
||
1,
|
||
"expected exactly 1 hit (the kind:9 control), got {} (kinds={kinds:?})",
|
||
result.hits.len(),
|
||
);
|
||
|
||
teardown(pool, &schema).await;
|
||
}
|
||
|
||
/// Tripwire: every Rust-side author-only kind MUST be excluded from
|
||
/// `search_tsv` at the storage layer.
|
||
///
|
||
/// The schema generated column hard-codes the privacy skip-set, while
|
||
/// `AUTHOR_ONLY_KINDS` is a Rust const. If a future author-only kind is added
|
||
/// without the matching schema migration, search would still spend FTS budget on
|
||
/// those private hits before the relay post-filter rejects them. Catch that
|
||
/// drift here by inserting one row per author-only kind and proving only the
|
||
/// public kind:9 control is searchable.
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn author_only_kinds_are_storage_level_unsearchable() {
|
||
let (pool, schema) = setup().await;
|
||
|
||
let c = mk_community(&pool, "author-only-tripwire.example").await;
|
||
let token = "authoronly_tripwire_marker_qwerty";
|
||
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
rand_bytes32(),
|
||
9,
|
||
&format!("public control — {token}"),
|
||
None,
|
||
1_700_000_000,
|
||
)
|
||
.await;
|
||
|
||
for (i, &kind) in AUTHOR_ONLY_KINDS.iter().enumerate() {
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
rand_bytes32(),
|
||
kind as i32,
|
||
&format!("author-only kind:{kind} — {token}"),
|
||
None,
|
||
1_700_000_100 + i as i64,
|
||
)
|
||
.await;
|
||
}
|
||
|
||
let svc = SearchService::new(pool.clone());
|
||
let result = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: token.into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: None,
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 100,
|
||
mode: buzz_search::SearchMode::FullText,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.expect("search ok");
|
||
|
||
let kinds: Vec<i32> = result.hits.iter().map(|h| h.kind).collect();
|
||
assert!(
|
||
kinds.contains(&9),
|
||
"kind:9 control row MUST be searchable, got kinds={kinds:?}",
|
||
);
|
||
|
||
for &kind in AUTHOR_ONLY_KINDS {
|
||
assert!(
|
||
!kinds.contains(&(kind as i32)),
|
||
"AUTHOR_ONLY kind:{kind} MUST NOT be searchable — \
|
||
schema skip-set is missing this kind. AUTHOR_ONLY_KINDS={AUTHOR_ONLY_KINDS:?}, \
|
||
hits={kinds:?}",
|
||
);
|
||
}
|
||
|
||
assert_eq!(
|
||
result.hits.len(),
|
||
1,
|
||
"expected exactly 1 hit (the kind:9 control), got {} (kinds={kinds:?})",
|
||
result.hits.len(),
|
||
);
|
||
|
||
teardown(pool, &schema).await;
|
||
}
|
||
|
||
/// Tripwire: every Rust-side `P_GATED_KINDS` entry that is *persistent* (not
|
||
/// in the ephemeral 20000–29999 range) MUST be excluded from `search_tsv` at
|
||
/// the storage layer.
|
||
///
|
||
/// L2 (the filter-level `#p` gate in `p_gated_filters_authorized`) prevents
|
||
/// reachable leaks today, but it is Rust logic — a future bug or new exempt
|
||
/// search entry point could surface tokenized content from these kinds. The
|
||
/// L1 NULL tsvector is the unbreakable backstop: `@@` mathematically cannot
|
||
/// match NULL. This test catches the drift where someone adds a persistent
|
||
/// kind to `P_GATED_KINDS` without the matching `schema/schema.sql` +
|
||
/// `migrations/0001_initial_schema.sql` skip-set update.
|
||
///
|
||
/// Ephemeral kinds (20000–29999) are skipped: they are never stored, so the
|
||
/// storage-layer defense does not apply to them regardless of the schema
|
||
/// CASE. `p_gated_filters_authorized` remains their sole defense by design.
|
||
///
|
||
/// Companion to `author_only_kinds_are_storage_level_unsearchable`: that test
|
||
/// covers `AUTHOR_ONLY_KINDS` drift; this one covers `P_GATED_KINDS`
|
||
/// persistent-subset drift. Together they tripwire both Rust-side privacy
|
||
/// constants against the schema literal.
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn p_gated_persistent_kinds_have_storage_null_tsvector() {
|
||
let (pool, schema) = setup().await;
|
||
|
||
let c = mk_community(&pool, "p-gated-tripwire.example").await;
|
||
let token = "pgated_tripwire_marker_qwerty";
|
||
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
rand_bytes32(),
|
||
9,
|
||
&format!("public control — {token}"),
|
||
None,
|
||
1_700_000_000,
|
||
)
|
||
.await;
|
||
|
||
let persistent: Vec<u32> = P_GATED_KINDS
|
||
.iter()
|
||
.copied()
|
||
.filter(|&k| !buzz_core::kind::is_ephemeral(k))
|
||
.collect();
|
||
assert!(
|
||
!persistent.is_empty(),
|
||
"P_GATED_KINDS must include at least one persistent kind for this \
|
||
test to be meaningful; got {P_GATED_KINDS:?}",
|
||
);
|
||
|
||
for (i, &kind) in persistent.iter().enumerate() {
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
rand_bytes32(),
|
||
kind as i32,
|
||
&format!("p-gated kind:{kind} — {token}"),
|
||
None,
|
||
1_700_000_100 + i as i64,
|
||
)
|
||
.await;
|
||
}
|
||
|
||
let svc = SearchService::new(pool.clone());
|
||
let result = svc
|
||
.search(&SearchQuery {
|
||
community: c,
|
||
q: token.into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: None,
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 100,
|
||
mode: buzz_search::SearchMode::FullText,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
})
|
||
.await
|
||
.expect("search ok");
|
||
|
||
let kinds: Vec<i32> = result.hits.iter().map(|h| h.kind).collect();
|
||
assert!(
|
||
kinds.contains(&9),
|
||
"kind:9 control row MUST be searchable, got kinds={kinds:?}",
|
||
);
|
||
|
||
for &kind in &persistent {
|
||
assert!(
|
||
!kinds.contains(&(kind as i32)),
|
||
"P_GATED persistent kind:{kind} MUST NOT be searchable — \
|
||
schema NULL tsvector skip-set is missing this kind. Defense \
|
||
reduces to L2 (filter-level `#p` gate) alone. \
|
||
P_GATED_KINDS={P_GATED_KINDS:?}, hits={kinds:?}",
|
||
);
|
||
}
|
||
|
||
assert_eq!(
|
||
result.hits.len(),
|
||
1,
|
||
"expected exactly 1 hit (the kind:9 control), got {} (kinds={kinds:?})",
|
||
result.hits.len(),
|
||
);
|
||
|
||
teardown(pool, &schema).await;
|
||
}
|
||
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn artifact_search_uses_only_live_current_head() {
|
||
let (pool, schema) = setup().await;
|
||
let community = mk_community(&pool, "artifact-search.example").await;
|
||
let pk = rand_bytes32();
|
||
let old = rand_bytes32();
|
||
let head = rand_bytes32();
|
||
let artifact = Uuid::new_v4();
|
||
insert_event(
|
||
&pool,
|
||
community,
|
||
old,
|
||
pk,
|
||
45010,
|
||
"artifactneedle old",
|
||
None,
|
||
1700000000,
|
||
)
|
||
.await;
|
||
insert_event(
|
||
&pool,
|
||
community,
|
||
head,
|
||
pk,
|
||
45010,
|
||
"artifactneedle current",
|
||
None,
|
||
1700000001,
|
||
)
|
||
.await;
|
||
sqlx::query("INSERT INTO artifact_heads(community_id,artifact_id,event_id,channel_id,artifact_type) VALUES($1,$2,$3,$4,'buzz.task')")
|
||
.bind(community.as_uuid()).bind(artifact).bind(head.as_slice()).bind(Uuid::new_v4()).execute(&pool).await.unwrap();
|
||
let svc = SearchService::new(pool.clone());
|
||
let query = SearchQuery {
|
||
community,
|
||
q: "artifactneedle".into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: Some(vec![45010]),
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::FullText,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
};
|
||
let result = svc.search(&query).await.unwrap();
|
||
assert_eq!(result.hits.len(), 1);
|
||
assert_eq!(result.hits[0].event_id, head);
|
||
sqlx::query("UPDATE artifact_heads SET deleted=true WHERE community_id=$1 AND artifact_id=$2")
|
||
.bind(community.as_uuid())
|
||
.bind(artifact)
|
||
.execute(&pool)
|
||
.await
|
||
.unwrap();
|
||
let result = svc.search(&query).await.unwrap();
|
||
assert!(result.hits.is_empty());
|
||
teardown(pool, &schema).await;
|
||
}
|
||
|
||
/// Insert a current kind:0 profile; returns its event id.
|
||
async fn put_profile(pool: &PgPool, community: CommunityId, content: &str, at: i64) -> [u8; 32] {
|
||
let id = rand_bytes32();
|
||
insert_event(pool, community, id, rand_bytes32(), 0, content, None, at).await;
|
||
id
|
||
}
|
||
|
||
async fn profile_names(
|
||
pool: &PgPool,
|
||
community: CommunityId,
|
||
q: &str,
|
||
page: u32,
|
||
per_page: u32,
|
||
) -> Vec<[u8; 32]> {
|
||
SearchService::new(pool.clone())
|
||
.search_profile_names(community, q, None, page, per_page)
|
||
.await
|
||
.expect("profile name search")
|
||
.hits
|
||
.into_iter()
|
||
.map(|hit| hit.event_id)
|
||
.collect()
|
||
}
|
||
|
||
/// Profile name search keeps the NIP-50 prefix rule (whole words in any
|
||
/// order, last word a prefix) but only `name` and `display_name` count. An
|
||
/// equal name ranks first, then a name prefix, then a later word prefix, then
|
||
/// other matches; ties stay newest first.
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn profile_names_match_name_words_only() {
|
||
let (pool, schema) = setup().await;
|
||
let c = mk_community(&pool, "profile-names.example").await;
|
||
let other = mk_community(&pool, "profile-names-other.example").await;
|
||
|
||
let exact = put_profile(
|
||
&pool,
|
||
c,
|
||
r#"{"display_name":"Wes","picture":"data:image/png;base64,P7+Wes+dYL"}"#,
|
||
1_700_000_000,
|
||
)
|
||
.await;
|
||
let westie_old = put_profile(&pool, c, r#"{"name":"Westie"}"#, 1_700_000_010).await;
|
||
let westie_new = put_profile(&pool, c, r#"{"display_name":"Westie"}"#, 1_700_000_020).await;
|
||
let word = put_profile(
|
||
&pool,
|
||
c,
|
||
r#"{"name":"carl","display_name":"Carl (borrowed from Wes)"}"#,
|
||
1_700_000_030,
|
||
)
|
||
.await;
|
||
let full = put_profile(
|
||
&pool,
|
||
c,
|
||
r#"{"name":"wjohnson","display_name":"Wes Johnson"}"#,
|
||
1_700_000_040,
|
||
)
|
||
.await;
|
||
// Words split across the two fields still match, as in NIP-50 search.
|
||
let split = put_profile(
|
||
&pool,
|
||
c,
|
||
r#"{"name":"johnson","display_name":"Wes"}"#,
|
||
1_700_000_050,
|
||
)
|
||
.await;
|
||
// A word that only contains the query is not a match: same as NIP-50.
|
||
put_profile(&pool, c, r#"{"name":"Dwes"}"#, 1_700_000_060).await;
|
||
|
||
for (i, noise) in [
|
||
r#"{"display_name":"Blueberry","picture":"data:image/png;base64,P7+Wes+dYL"}"#,
|
||
r#"{"display_name":"Pollen","about":"Wes Johnson's bestie"}"#,
|
||
r#"{"display_name":"Pollen","nip05":"wes@example.com"}"#,
|
||
r#"{"display_name":"Pollen","banner":"https://wes.example/b.png"}"#,
|
||
r#"{"name":["wes"],"display_name":{"text":"wes"}}"#,
|
||
r#"not json wes"#,
|
||
r#"["wes"]"#,
|
||
]
|
||
.into_iter()
|
||
.enumerate()
|
||
{
|
||
put_profile(&pool, c, noise, 1_700_000_100 + i as i64).await;
|
||
}
|
||
// A replaced profile (soft-deleted) no longer matches.
|
||
let replaced = put_profile(&pool, c, r#"{"name":"wes"}"#, 1_699_000_000).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 profile");
|
||
put_profile(&pool, other, r#"{"name":"Wes"}"#, 1_700_000_000).await;
|
||
|
||
// Tier 0: exact (Wes; "Wes" display name of `split`), newest first.
|
||
// Tier 1: name prefix (Wes Johnson, Westie x2). Tier 2: later word (Carl).
|
||
let all = vec![split, exact, full, westie_new, westie_old, word];
|
||
assert_eq!(profile_names(&pool, c, "WES", 1, 100).await, all);
|
||
// Pages split the same order without overlap or gaps.
|
||
assert_eq!(profile_names(&pool, c, "wes", 1, 4).await, all[..4]);
|
||
assert_eq!(profile_names(&pool, c, "wes", 2, 4).await, all[4..]);
|
||
assert!(profile_names(&pool, c, "wes", 3, 4).await.is_empty());
|
||
|
||
// Word order does not matter; completed words must be whole words.
|
||
assert_eq!(
|
||
profile_names(&pool, c, "johnson wes", 1, 100).await,
|
||
vec![split, full]
|
||
);
|
||
assert_eq!(
|
||
profile_names(&pool, c, "wes jo", 1, 100).await,
|
||
vec![full, split]
|
||
);
|
||
assert!(profile_names(&pool, c, "estie", 1, 100).await.is_empty());
|
||
assert!(profile_names(&pool, c, "bestie", 1, 100).await.is_empty());
|
||
assert!(profile_names(&pool, c, "example", 1, 100).await.is_empty());
|
||
assert!(profile_names(&pool, c, "%", 1, 100).await.is_empty());
|
||
assert!(profile_names(&pool, c, " ", 1, 100).await.is_empty());
|
||
|
||
// The query and the names are normalized by the same SQL, so a non-ASCII
|
||
// exact name is tier 0 whatever the database collation lower-cases.
|
||
let emile = put_profile(&pool, c, r#"{"name":"Émile"}"#, 1_700_000_200).await;
|
||
let emilee = put_profile(&pool, c, r#"{"name":"Émilee"}"#, 1_700_000_210).await;
|
||
assert_eq!(
|
||
profile_names(&pool, c, "Émile", 1, 100).await,
|
||
vec![emile, emilee]
|
||
);
|
||
|
||
teardown(pool, &schema).await;
|
||
}
|
||
|
||
/// Stored kind:0 content is whatever the signer sent. Content that cannot
|
||
/// become jsonb must not fail the query for everyone else.
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn profile_names_survive_unparsable_content() {
|
||
let (pool, schema) = setup().await;
|
||
let c = mk_community(&pool, "profile-names-unparsable.example").await;
|
||
|
||
let plain = put_profile(&pool, c, r#"{"name":"Wes"}"#, 1_700_000_000).await;
|
||
|
||
// Valid JSON objects that cannot become jsonb. Each one names Wes; it is
|
||
// not findable, but it must not break the search.
|
||
let deep = format!(
|
||
r#"{{"name":"Wes","nested":{}{}}}"#,
|
||
"[".repeat(60_000),
|
||
"]".repeat(60_000)
|
||
);
|
||
for (i, poison) in [
|
||
r#"{"name":"Wes","about":"\u0000"}"#,
|
||
r#"{"name":"Wes","about":"\ud800"}"#,
|
||
r#"{"name":"Wes","lud16":1e1000000}"#,
|
||
deep.as_str(),
|
||
]
|
||
.into_iter()
|
||
.enumerate()
|
||
{
|
||
put_profile(&pool, c, poison, 1_700_000_100 + i as i64).await;
|
||
}
|
||
|
||
assert_eq!(profile_names(&pool, c, "wes", 1, 100).await, vec![plain]);
|
||
|
||
// The nesting bound counts brackets, so content at the bound (4096,
|
||
// counting the object's own `{`) still has names even when the other
|
||
// brackets sit inside a string.
|
||
let bracketed = put_profile(
|
||
&pool,
|
||
c,
|
||
&format!(r#"{{"name":"Ada","about":"{}"}}"#, "[".repeat(4095)),
|
||
1_700_000_200,
|
||
)
|
||
.await;
|
||
assert_eq!(
|
||
profile_names(&pool, c, "ada", 1, 100).await,
|
||
vec![bracketed]
|
||
);
|
||
|
||
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());
|
||
buzz_search::pool_options()
|
||
.max_connections(4)
|
||
.connect(&url)
|
||
.await
|
||
.expect("connect search pool")
|
||
}
|
||
|
||
/// 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. 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 'SELECT id, kind, pubkey, channel_id, %ts_rank_cd(%' \
|
||
OR query LIKE 'WITH search_query AS (SELECT %')",
|
||
)
|
||
.fetch_one(pool)
|
||
.await
|
||
.expect("count waiting searches")
|
||
}
|
||
|
||
/// Poll until [`waiting_searches`] is `want`, failing after five seconds.
|
||
async fn await_waiting_searches(pool: &PgPool, want: i64, why: &str) {
|
||
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
|
||
loop {
|
||
let got = waiting_searches(pool).await;
|
||
if got == want {
|
||
return;
|
||
}
|
||
assert!(
|
||
tokio::time::Instant::now() < deadline,
|
||
"{why}: {got} waiting searches"
|
||
);
|
||
tokio::time::sleep(std::time::Duration::from_millis(25)).await;
|
||
}
|
||
}
|
||
|
||
/// Hold `start()`'s search on a table lock, drop it while it waits, and
|
||
/// require that its statement stops; then run it again unlocked.
|
||
async fn abandon_under_lock<F, Fut>(pool: &PgPool, name: &str, start: F)
|
||
where
|
||
F: Fn() -> Fut,
|
||
Fut: std::future::Future<Output = Result<buzz_search::SearchResult, buzz_search::SearchError>>,
|
||
{
|
||
let mut lock = pool.begin().await.expect("begin lock");
|
||
sqlx::query("LOCK TABLE events IN ACCESS EXCLUSIVE MODE")
|
||
.execute(&mut *lock)
|
||
.await
|
||
.expect("lock events");
|
||
// Boxed so `drop` drops the search itself, not a pinned reference to it.
|
||
let mut running = Box::pin(start());
|
||
tokio::select! {
|
||
result = &mut running => panic!("{name} finished under the lock: {result:?}"),
|
||
() = await_waiting_searches(pool, 1, "search never reached the lock") => {}
|
||
}
|
||
drop(running);
|
||
await_waiting_searches(pool, 0, &format!("abandoned {name} still running")).await;
|
||
lock.rollback().await.expect("release lock");
|
||
// The cancelled connection is usable again.
|
||
start()
|
||
.await
|
||
.unwrap_or_else(|e| panic!("{name} after cancel: {e}"));
|
||
}
|
||
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn search_pool_bounds_every_statement() {
|
||
let (pool, schema) = setup().await;
|
||
let search = search_pool().await;
|
||
let mut conns = Vec::new();
|
||
for _ in 0..2 {
|
||
let mut conn = search.acquire().await.expect("search connection");
|
||
let timeout: String = sqlx::query_scalar("SHOW statement_timeout")
|
||
.fetch_one(&mut *conn)
|
||
.await
|
||
.expect("read statement_timeout");
|
||
assert_eq!(timeout, "20s", "every search connection is bounded");
|
||
conns.push(conn);
|
||
}
|
||
drop(conns);
|
||
search.close().await;
|
||
teardown(pool, &schema).await;
|
||
}
|
||
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn abandoned_search_cancels_its_statement() {
|
||
let (pool, schema) = setup().await;
|
||
let c = mk_community(&pool, "abandoned-search.example").await;
|
||
insert_event(
|
||
&pool,
|
||
c,
|
||
rand_bytes32(),
|
||
rand_bytes32(),
|
||
9,
|
||
"project plan",
|
||
None,
|
||
1_700_000_000,
|
||
)
|
||
.await;
|
||
let svc = SearchService::new(search_pool().await);
|
||
let message_query = SearchQuery {
|
||
community: c,
|
||
q: "project".into(),
|
||
channel_scope: ChannelScope::Any,
|
||
kinds: Some(vec![9]),
|
||
authors: None,
|
||
since: None,
|
||
until: None,
|
||
page: 1,
|
||
per_page: 10,
|
||
mode: buzz_search::SearchMode::Prefix,
|
||
sort: buzz_search::SearchSort::Relevance,
|
||
};
|
||
abandon_under_lock(&pool, "search", || svc.search(&message_query)).await;
|
||
abandon_under_lock(&pool, "search_profile_names", || {
|
||
svc.search_profile_names(c, "proj", None, 1, 10)
|
||
})
|
||
.await;
|
||
teardown(pool, &schema).await;
|
||
}
|
||
|
||
#[tokio::test]
|
||
#[ignore = "requires Postgres"]
|
||
async fn cancel_signals_only_the_named_backend() {
|
||
let (pool, schema) = setup().await;
|
||
let mut victim = pool.acquire().await.expect("victim connection");
|
||
let (pid, started_us): (i32, i64) = sqlx::query_as(
|
||
"SELECT pid, (EXTRACT(EPOCH FROM backend_start) * 1000000)::bigint \
|
||
FROM pg_stat_activity WHERE pid = pg_backend_pid()",
|
||
)
|
||
.fetch_one(&mut *victim)
|
||
.await
|
||
.expect("victim backend");
|
||
let sleeping = tokio::spawn(async move {
|
||
sqlx::query("SELECT pg_sleep(30)")
|
||
.execute(&mut *victim)
|
||
.await
|
||
.map(|_| ())
|
||
});
|
||
// Signal only once the victim's statement is running: a cancel that
|
||
// lands first is a no-op, and the named cancel below would then pass
|
||
// without stopping anything.
|
||
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
|
||
loop {
|
||
let running: bool = sqlx::query_scalar(
|
||
"SELECT EXISTS (SELECT 1 FROM pg_stat_activity \
|
||
WHERE pid = $1 AND state = 'active' AND query LIKE '%pg_sleep(30)%')",
|
||
)
|
||
.bind(pid)
|
||
.fetch_one(&pool)
|
||
.await
|
||
.expect("victim activity");
|
||
if running {
|
||
break;
|
||
}
|
||
assert!(
|
||
tokio::time::Instant::now() < deadline,
|
||
"victim statement never started"
|
||
);
|
||
tokio::time::sleep(std::time::Duration::from_millis(25)).await;
|
||
}
|
||
|
||
// A same-pid backend that started at another time (another replica, or a
|
||
// reused pid) is not signalled.
|
||
let other: Option<bool> = sqlx::query_scalar(buzz_search::cancel::CANCEL_SQL)
|
||
.bind(pid)
|
||
.bind(started_us + 1)
|
||
.fetch_optional(&pool)
|
||
.await
|
||
.expect("cancel other backend");
|
||
assert_eq!(other, None);
|
||
tokio::time::sleep(std::time::Duration::from_millis(200)).await;
|
||
assert!(
|
||
!sleeping.is_finished(),
|
||
"an unmatched cancel must not stop the statement"
|
||
);
|
||
|
||
let named: Option<bool> = sqlx::query_scalar(buzz_search::cancel::CANCEL_SQL)
|
||
.bind(pid)
|
||
.bind(started_us)
|
||
.fetch_optional(&pool)
|
||
.await
|
||
.expect("cancel named backend");
|
||
assert_eq!(named, Some(true));
|
||
let err = tokio::time::timeout(std::time::Duration::from_secs(5), sleeping)
|
||
.await
|
||
.expect("cancelled statement ends")
|
||
.expect("join")
|
||
.expect_err("statement was cancelled");
|
||
assert_eq!(
|
||
err.as_database_error().and_then(|e| e.code()).as_deref(),
|
||
Some("57014")
|
||
);
|
||
teardown(pool, &schema).await;
|
||
}
|