feat(nip77): add negentropy reconciliation core (#809)

* feat(nip77): add negentropy set reconciliation core

Signed-off-by: Priyanshubhartistm <bhartipriyanshustm@gmail.com>

* feat(nip77): add repository query for reconciliation items

Signed-off-by: Priyanshubhartistm <bhartipriyanshustm@gmail.com>

* test(nip77): cover negentropy core and reconciliation query

Signed-off-by: Priyanshubhartistm <bhartipriyanshustm@gmail.com>

* fix(nip77): address review findings in negentropy core

Signed-off-by: Priyanshubhartistm <bhartipriyanshustm@gmail.com>

---------

Signed-off-by: Priyanshubhartistm <bhartipriyanshustm@gmail.com>
Co-authored-by: Ricardo Cabral <me@ricardocabral.io>

Author: Priyanshubhartistm

.changeset/nip77-negentropy-core.md

@@ -0,0 +1,5 @@
+---
+"nostream": minor
+---
+
+feat(nip77): add negentropy set-reconciliation core and a sorted event id query

.knip.json

@@ -15,6 +15,7 @@
   ],
   "ignore": [
     ".nostr/**",
+    "src/utils/negentropy.ts",
     "src/utils/relay-probe/**"
   ],
   "commitlint": false,

src/@types/repositories.ts

@@ -31,6 +31,8 @@ export interface EventPurgeCounts {
 
 export type ExposedPromiseKeys = 'then' | 'catch' | 'finally'
 
+export type ReconciliationItem = Pick<DBEvent, 'event_created_at' | 'event_id'>
+
 export interface IQueryResult<T> extends Pick<Promise<T>, keyof Promise<T> & ExposedPromiseKeys> {
   stream(options?: Record<string, any>): PassThrough & AsyncIterable<T>
 }
@@ -42,6 +44,7 @@ export interface IEventRepository {
   upsertMany(events: Event[]): Promise<number>
   findByFilters(filters: SubscriptionFilter[]): IQueryResult<DBEvent[]>
   countByFilters(filters: SubscriptionFilter[]): Promise<number>
+  findReconciliationItems(filter: SubscriptionFilter): IQueryResult<ReconciliationItem[]>
   deleteByPubkeyAndIds(pubkey: Pubkey, ids: EventId[]): Promise<number>
   deleteByPubkeyExceptKinds(pubkey: Pubkey, excludedKinds: number[]): Promise<number>
   hasActiveRequestToVanish(pubkey: Pubkey): Promise<boolean>

src/repositories/event-repository.ts

@@ -23,7 +23,13 @@ import {
 } from '../constants/base'
 import { DatabaseClient, EventId } from '../@types/base'
 import { DBEvent, Event } from '../@types/event'
-import { EventPurgeCounts, EventRetentionOptions, IEventRepository, IQueryResult } from '../@types/repositories'
+import {
+  EventPurgeCounts,
+  EventRetentionOptions,
+  IEventRepository,
+  IQueryResult,
+  ReconciliationItem,
+} from '../@types/repositories'
 import { toBuffer, toJSON } from '../utils/transform'
 import { createLogger } from '../factories/logger-factory'
 import { isGenericTagQuery, isGeohashPrefixCriterion, stripGeohashPrefixWildcard } from '../utils/filter'
@@ -128,6 +134,31 @@ export class EventRepository implements IEventRepository {
     return Number(result?.count ?? 0)
   }
 
+  /**
+   * Every live event matching the filter as (created_at, id), oldest first, for NIP-77 set
+   * reconciliation. The filter's limit is ignored on purpose: reconciliation compares whole sets, so
+   * truncating one side would report the cut-off events as missing. Deleted and expired events are
+   * excluded because they must never be advertised to a peer.
+   */
+  public findReconciliationItems(filter: SubscriptionFilter): IQueryResult<ReconciliationItem[]> {
+    logger('querying reconciliation items for %o', filter)
+
+    const now = Math.floor(Date.now() / 1000)
+    const builder = this.readReplicaDbClient<DBEvent>('events').select('events.event_created_at', 'events.event_id')
+
+    this.applyFilterConditions(builder, filter)
+
+    builder
+      .whereNull('events.deleted_at')
+      .andWhere((bd) => {
+        bd.whereNull('events.expires_at').orWhere('events.expires_at', '>', now)
+      })
+      .orderBy('events.event_created_at', 'asc')
+      .orderBy('events.event_id', 'asc')
+
+    return builder as unknown as IQueryResult<ReconciliationItem[]>
+  }
+
   // Wraps every query so each keeps its own ORDER BY and LIMIT. first.union(rest) would put the
   // first query's ORDER BY and LIMIT after the UNION, applying them to the whole result.
   private unionFilterQueries(queries: any[]): any {