fix: de-duplicate events returned by generic tag-filter subscriptions

.changeset/duplicate-events-on-tag-filter-queries.md

@@ -0,0 +1,15 @@
+---
+"nostream": patch
+---
+
+fix: de-duplicate events returned by generic tag-filter subscriptions
+
+`EventRepository.findByFilters()` left-joins `event_tags` for generic tag filters
+(`#e`, `#p`, etc.) without deduplicating the result. An event matching more than one
+tag row for the same filter (e.g. `{"#p": ["a", "b"]}` matching an event tagged with
+both) was returned once per matching `event_tags` row, so subscribers received the
+same `EVENT` message multiple times. The query now selects `DISTINCT events.*` for
+tag-filtered queries so each stored event is returned at most once. This also covers
+generic tag filters combined with a NIP-50 `search` term (e.g.
+`{"search": "...", "#p": ["a", "b"]}`), which take the search branch and are now
+de-duplicated as well.

src/repositories/event-repository.ts

@@ -73,24 +73,25 @@ export class EventRepository implements IEventRepository {
  const maxLen = nip50Settings?.nip50?.maxQueryLength ?? DEFAULT_MAX_SEARCH_QUERY_LENGTH
  const searchQuery = currentFilter.search.trim().slice(0, maxLen)
  const limit = typeof currentFilter.limit === 'number' ? currentFilter.limit : DEFAULT_FILTER_LIMIT
- builder
- .select(
- this.readReplicaDbClient.raw(
- 'events.*, ts_rank(to_tsvector(?::regconfig, event_content), plainto_tsquery(?::regconfig, ?)) AS search_rank',
- [tsConfig, tsConfig, searchQuery],
- ),
- )
- .limit(limit)
- .orderBy('search_rank', 'DESC')
- .orderBy('event_id', 'asc')
+ const searchSelection = this.readReplicaDbClient.raw(
+ 'events.*, ts_rank(to_tsvector(?::regconfig, event_content), plainto_tsquery(?::regconfig, ?)) AS search_rank',
+ [tsConfig, tsConfig, searchQuery],
+ )
+ // De-duplicate rows multiplied by the event_tags left join when search is combined with a generic tag filter
+ if (isTagQuery) {
+ builder.distinct(searchSelection)
+ } else {
+ builder.select(searchSelection)
+ }
+ builder.limit(limit).orderBy('search_rank', 'DESC').orderBy('event_id', 'asc')
  } else if (typeof currentFilter.limit === 'number') {
  builder.limit(currentFilter.limit).orderBy('event_created_at', 'DESC').orderBy('event_id', 'asc')
  } else {
  builder.limit(DEFAULT_FILTER_LIMIT).orderBy('event_created_at', 'asc').orderBy('event_id', 'asc')
  }
 
  if (isTagQuery && !isSearchQuery) {
- builder.select('events.*')
+ builder.distinct('events.*')
  }
 
  return builder

test/unit/repositories/event-repository.spec.ts

@@ -295,7 +295,7 @@ describe('EventRepository', () => {
  const query = repository.findByFilters(filters).toString()
 
  expect(query).to.equal(
- 'select "events".* from "events" left join "event_tags" on "events"."event_id" = "event_tags"."event_id" where (1 = 0) order by "event_created_at" asc, "event_id" asc limit 500',
+ 'select distinct "events".* from "events" left join "event_tags" on "events"."event_id" = "event_tags"."event_id" where (1 = 0) order by "event_created_at" asc, "event_id" asc limit 500',
  )
  })
 
@@ -305,7 +305,7 @@ describe('EventRepository', () => {
  const query = repository.findByFilters(filters).toString()
 
  expect(query).to.equal(
- 'select "events".* from "events" left join "event_tags" on "events"."event_id" = "event_tags"."event_id" where (event_tags.tag_name = \'e\' AND event_tags.tag_value = \'aaaaaa\') order by "event_created_at" asc, "event_id" asc limit 500',
+ 'select distinct "events".* from "events" left join "event_tags" on "events"."event_id" = "event_tags"."event_id" where (event_tags.tag_name = \'e\' AND event_tags.tag_value = \'aaaaaa\') order by "event_created_at" asc, "event_id" asc limit 500',
  )
  })
 
@@ -315,7 +315,7 @@ describe('EventRepository', () => {
  const query = repository.findByFilters(filters).toString()
 
  expect(query).to.equal(
- 'select "events".* from "events" left join "event_tags" on "events"."event_id" = "event_tags"."event_id" where (event_tags.tag_name = \'e\' AND event_tags.tag_value = \'aaaaaa\' or event_tags.tag_name = \'e\' AND event_tags.tag_value = \'bbbbbb\') order by "event_created_at" asc, "event_id" asc limit 500',
+ 'select distinct "events".* from "events" left join "event_tags" on "events"."event_id" = "event_tags"."event_id" where (event_tags.tag_name = \'e\' AND event_tags.tag_value = \'aaaaaa\' or event_tags.tag_name = \'e\' AND event_tags.tag_value = \'bbbbbb\') order by "event_created_at" asc, "event_id" asc limit 500',
  )
  })
  })
@@ -327,7 +327,7 @@ describe('EventRepository', () => {
  const query = repository.findByFilters(filters).toString()
 
  expect(query).to.equal(
- 'select "events".* from "events" left join "event_tags" on "events"."event_id" = "event_tags"."event_id" where (event_tags.tag_name = \'g\' AND event_tags.tag_value LIKE \'u4pruyd%\') order by "event_created_at" asc, "event_id" asc limit 500',
+ 'select distinct "events".* from "events" left join "event_tags" on "events"."event_id" = "event_tags"."event_id" where (event_tags.tag_name = \'g\' AND event_tags.tag_value LIKE \'u4pruyd%\') order by "event_created_at" asc, "event_id" asc limit 500',
  )
  })
 
@@ -337,7 +337,7 @@ describe('EventRepository', () => {
  const query = repository.findByFilters(filters).toString()
 
  expect(query).to.equal(
- 'select "events".* from "events" left join "event_tags" on "events"."event_id" = "event_tags"."event_id" where (event_tags.tag_name = \'g\' AND event_tags.tag_value = \'u4pruyd\') order by "event_created_at" asc, "event_id" asc limit 500',
+ 'select distinct "events".* from "events" left join "event_tags" on "events"."event_id" = "event_tags"."event_id" where (event_tags.tag_name = \'g\' AND event_tags.tag_value = \'u4pruyd\') order by "event_created_at" asc, "event_id" asc limit 500',
  )
  })
  })
@@ -349,7 +349,7 @@ describe('EventRepository', () => {
  const query = repository.findByFilters(filters).toString()
 
  expect(query).to.equal(
- 'select "events".* from "events" left join "event_tags" on "events"."event_id" = "event_tags"."event_id" where (1 = 0) order by "event_created_at" asc, "event_id" asc limit 500',
+ 'select distinct "events".* from "events" left join "event_tags" on "events"."event_id" = "event_tags"."event_id" where (1 = 0) order by "event_created_at" asc, "event_id" asc limit 500',
  )
  })
 
@@ -359,7 +359,7 @@ describe('EventRepository', () => {
  const query = repository.findByFilters(filters).toString()
 
  expect(query).to.equal(
- 'select "events".* from "events" left join "event_tags" on "events"."event_id" = "event_tags"."event_id" where (event_tags.tag_name = \'p\' AND event_tags.tag_value = \'aaaaaa\') order by "event_created_at" asc, "event_id" asc limit 500',
+ 'select distinct "events".* from "events" left join "event_tags" on "events"."event_id" = "event_tags"."event_id" where (event_tags.tag_name = \'p\' AND event_tags.tag_value = \'aaaaaa\') order by "event_created_at" asc, "event_id" asc limit 500',
  )
  })
 
@@ -369,7 +369,7 @@ describe('EventRepository', () => {
  const query = repository.findByFilters(filters).toString()
 
  expect(query).to.equal(
- 'select "events".* from "events" left join "event_tags" on "events"."event_id" = "event_tags"."event_id" where (event_tags.tag_name = \'p\' AND event_tags.tag_value = \'aaaaaa\' or event_tags.tag_name = \'p\' AND event_tags.tag_value = \'bbbbbb\') order by "event_created_at" asc, "event_id" asc limit 500',
+ 'select distinct "events".* from "events" left join "event_tags" on "events"."event_id" = "event_tags"."event_id" where (event_tags.tag_name = \'p\' AND event_tags.tag_value = \'aaaaaa\' or event_tags.tag_name = \'p\' AND event_tags.tag_value = \'bbbbbb\') order by "event_created_at" asc, "event_id" asc limit 500',
  )
  })
  })
@@ -381,7 +381,7 @@ describe('EventRepository', () => {
  const query = repository.findByFilters(filters).toString()
 
  expect(query).to.equal(
- 'select "events".* from "events" left join "event_tags" on "events"."event_id" = "event_tags"."event_id" where (1 = 0) order by "event_created_at" asc, "event_id" asc limit 500',
+ 'select distinct "events".* from "events" left join "event_tags" on "events"."event_id" = "event_tags"."event_id" where (1 = 0) order by "event_created_at" asc, "event_id" asc limit 500',
  )
  })
 
@@ -391,7 +391,7 @@ describe('EventRepository', () => {
  const query = repository.findByFilters(filters).toString()
 
  expect(query).to.equal(
- 'select "events".* from "events" left join "event_tags" on "events"."event_id" = "event_tags"."event_id" where (event_tags.tag_name = \'r\' AND event_tags.tag_value = \'aaaaaa\') order by "event_created_at" asc, "event_id" asc limit 500',
+ 'select distinct "events".* from "events" left join "event_tags" on "events"."event_id" = "event_tags"."event_id" where (event_tags.tag_name = \'r\' AND event_tags.tag_value = \'aaaaaa\') order by "event_created_at" asc, "event_id" asc limit 500',
  )
  })
 
@@ -401,7 +401,7 @@ describe('EventRepository', () => {
  const query = repository.findByFilters(filters).toString()
 
  expect(query).to.equal(
- 'select "events".* from "events" left join "event_tags" on "events"."event_id" = "event_tags"."event_id" where (event_tags.tag_name = \'r\' AND event_tags.tag_value = \'aaaaaa\' or event_tags.tag_name = \'r\' AND event_tags.tag_value = \'bbbbbb\') order by "event_created_at" asc, "event_id" asc limit 500',
+ 'select distinct "events".* from "events" left join "event_tags" on "events"."event_id" = "event_tags"."event_id" where (event_tags.tag_name = \'r\' AND event_tags.tag_value = \'aaaaaa\' or event_tags.tag_name = \'r\' AND event_tags.tag_value = \'bbbbbb\') order by "event_created_at" asc, "event_id" asc limit 500',
  )
  })
  })
@@ -413,7 +413,7 @@ describe('EventRepository', () => {
  const query = repository.findByFilters(filters).toString()
 
  expect(query).to.equal(
- 'select "events".* from "events" left join "event_tags" on "events"."event_id" = "event_tags"."event_id" where (event_tags.tag_name = \'d\' AND event_tags.tag_value = \'\') order by "event_created_at" asc, "event_id" asc limit 500',
+ 'select distinct "events".* from "events" left join "event_tags" on "events"."event_id" = "event_tags"."event_id" where (event_tags.tag_name = \'d\' AND event_tags.tag_value = \'\') order by "event_created_at" asc, "event_id" asc limit 500',
  )
  })
  })
@@ -501,6 +501,18 @@ describe('EventRepository', () => {
  expect(query).to.include('"event_kind" in (1)')
  })
 
+ it('de-duplicates results when search is combined with a generic tag filter', () => {
+ const filters = [{ search: 'bitcoin', '#p': ['a', 'b'] }]
+
+ const query = searchEnabledRepository.findByFilters(filters).toString()
+
+ expect(query).to.include('select distinct events.*')
+ expect(query).to.include('ts_rank(')
+ expect(query).to.include('left join "event_tags" on "events"."event_id" = "event_tags"."event_id"')
+ expect(query).to.include("plainto_tsquery('simple'::regconfig, 'bitcoin')")
+ expect(query).to.include("event_tags.tag_name = 'p'")
+ })
+
  it('ignores search filter when NIP-50 is disabled', () => {
  const disabledRepository = new EventRepository(dbClient, rrDbClient, () => ({
  nip50: { enabled: false },