fix(nip56): address post-merge review findings on report ingestion

.changeset/nip56-report-ingestion-gaps.md

@@ -0,0 +1,5 @@
+---
+"nostream": patch
+---
+
+fix(nip56): skip targetless report rows, record every p/e target, batch report inserts in one transaction, and warm the WoT graph at boot instead of blocking the first report on a cold-start rebuild

CONFIGURATION.md

@@ -215,7 +215,7 @@ The settings below are listed in alphabetical order by name. Please keep this ta
 | nip50.enabled                               | Enable or disable NIP-50 full-text search. Defaults to false. When enabled, clients can include a `search` field in REQ filters to perform text queries against event content. Requires the GIN full-text index migration. |
 | nip50.language                              | PostgreSQL text-search configuration name. Defaults to `simple` (language-agnostic tokenization). Set to `english`, `spanish`, etc. for stemming support. See [PostgreSQL text search configurations](https://www.postgresql.org/docs/current/textsearch-configuration.html). **Note:** The GIN index migration is built with the `simple` configuration. If you change this value, you must manually rebuild the index: `DROP INDEX CONCURRENTLY events_content_fts_idx; CREATE INDEX CONCURRENTLY events_content_fts_idx ON events USING gin (to_tsvector('<your_language>', event_content));` — otherwise the planner cannot use the index and queries fall back to sequential scans. |
 | nip50.maxQueryLength                        | Maximum length of the search query string. Queries exceeding this are truncated. Defaults to 256. |
-| nip56.enabled                               | Enable NIP-56 content reporting. When true, kind-1984 report events are stored and scored by the reporter's WoT distance from `wot.seedPubkey`. Defaults to false. |
+| nip56.enabled                               | Enable NIP-56 content reporting. When true, kind-1984 report events are stored and scored by the reporter's WoT distance from `wot.seedPubkey`. Defaults to false. If `wot.enabled` is false, every non-moderator report is still stored, but its WoT distance is always undefined, so it scores weight 0 and never becomes actionable — this matches the "record, never act" design, but is easy to misread as a misconfiguration. `reports` rows have no automatic retention/pruning: they can outlive the kind-1984 event that produced them (which is itself subject to normal event retention/deletion) and accumulate indefinitely; operators who want bounded growth need to prune the table themselves. |
 | nip56.trustedModerators                     | Pubkeys (hex) whose reports are always maximum-weight and actionable, regardless of WoT distance. Reports from any other pubkey are stored and weighted, but never trigger automatic actions on their own. Defaults to []. |
 | nip66.dnsCacheTtlSeconds                    | DNS cache TTL in seconds for repeated probe lookups of the same hostname. Defaults to 300. |
 | nip66.enabled                               | Enable NIP-66 relay monitoring. When true, starts a `relay-monitor` cluster worker that probes targets on an interval and stores the latest snapshot in Redis. Defaults to false. |

src/@types/repositories.ts

@@ -6,12 +6,12 @@ import { DBEvent, Event } from './event'
 import { CreateInviteCodeOptions, InviteCode } from './invite-code'
 import { Invoice } from './invoice'
 import { Nip05Verification } from './nip05'
+import { NotificationOutboxMessage, NotificationOutboxPayload } from './notification-outbox'
 import {
   NotificationDeliveryLogEntry,
   NotificationDeliveryStatus,
   OperatorNotificationChannelType,
 } from './operator-notifications'
-import { NotificationOutboxMessage, NotificationOutboxPayload } from './notification-outbox'
 import { Report } from './report'
 import { EventKindsRange } from './settings'
 import { SubscriptionFilter } from './subscription'
@@ -93,6 +93,8 @@ export interface IDvmJobRepository {
 
 export interface IReportRepository {
   create(report: Omit<Report, 'id' | 'createdAt'>): Promise<Report>
+  /** Inserts every row in a single transaction, so a mid-batch failure leaves none of them. */
+  createMany(reports: Omit<Report, 'id' | 'createdAt'>[]): Promise<Report[]>
   findByEventId(eventId: EventId): Promise<Report[]>
   findActionable(limit?: number): Promise<Report[]>
 }

src/@types/services.ts

@@ -1,5 +1,5 @@
-import { Invoice } from './invoice'
 import { Pubkey } from './base'
+import { Invoice } from './invoice'
 import { NotificationOutboxPayload } from './notification-outbox'
 
 export interface IMaintenanceService {
@@ -17,6 +17,12 @@ export interface IWotGraphService {
   isTrusted(pubkey: Pubkey): Promise<boolean>
   /** Applies a pubkey's current NIP-02 follow list to the graph. */
   updateFollowList(pubkey: Pubkey, follows: Pubkey[]): Promise<void>
+  /**
+   * Fire-and-forget: kicks off the initial graph build without waiting on it,
+   * so the first real getDistance() call after startup doesn't have to pay
+   * for a cold-start rebuild itself.
+   */
+  warmUp(): void
 }
 
 export interface IPaymentsService {
@@ -35,11 +41,7 @@ export interface NotificationDispatchContext {
 }
 
 export interface INotificationDispatcher {
-  dispatch(
-    eventType: string,
-    payload: NotificationOutboxPayload,
-    context?: NotificationDispatchContext,
-  ): Promise<void>
+  dispatch(eventType: string, payload: NotificationOutboxPayload, context?: NotificationDispatchContext): Promise<void>
 }
 
 export interface IOperatorNotificationService extends INotificationDispatcher {

src/factories/message-handler-factory.ts

@@ -1,4 +1,5 @@
 import { ICacheAdapter, IWebSocketAdapter } from '../@types/adapters'
+import { IncomingMessage, MessageType } from '../@types/messages'
 import {
   IDvmJobRepository,
   IEventRepository,
@@ -7,21 +8,20 @@ import {
   IReportRepository,
   IUserRepository,
 } from '../@types/repositories'
-import { IncomingMessage, MessageType } from '../@types/messages'
-import { createSettings } from './settings-factory'
+import { RedisAdapter } from '../adapters/redis-adapter'
+import { getCacheClient } from '../cache/client'
 import { AuthMessageHandler } from '../handlers/auth-message-handler'
 import { CountMessageHandler } from '../handlers/count-message-handler'
 import { EventMessageHandler } from '../handlers/event-message-handler'
-import { eventStrategyFactory } from './event-strategy-factory'
-import { getCacheClient } from '../cache/client'
-import { RedisAdapter } from '../adapters/redis-adapter'
-import { rateLimiterFactory } from './rate-limiter-factory'
 import { SubscribeMessageHandler } from '../handlers/subscribe-message-handler'
 import { UnsubscribeMessageHandler } from '../handlers/unsubscribe-message-handler'
+import { eventStrategyFactory } from './event-strategy-factory'
+import { rateLimiterFactory } from './rate-limiter-factory'
+import { createSettings } from './settings-factory'
 import { wotGraphServiceFactory } from './wot-graph-service-factory'
 
 let cacheAdapter: ICacheAdapter | undefined = undefined
-const getCache = (): ICacheAdapter => {
+export const getCache = (): ICacheAdapter => {
   if (!cacheAdapter) {
     cacheAdapter = new RedisAdapter(getCacheClient())
   }

src/factories/worker-factory.ts

@@ -1,21 +1,22 @@
-import { is, path, pathSatisfies } from 'ramda'
 import http from 'http'
 import process from 'process'
+import { is, path, pathSatisfies } from 'ramda'
 import { WebSocketServer } from 'ws'
-
-import { getMasterDbClient, getReadReplicaDbClient } from '../database/client'
+import { WebSocketServerAdapter } from '../adapters/web-socket-server-adapter'
 import { AppWorker } from '../app/worker'
-import { createLogger } from './logger-factory'
+import { getMasterDbClient, getReadReplicaDbClient } from '../database/client'
 import { createSettings } from '../factories/settings-factory'
-import { createWebApp } from './web-app-factory'
 import { DvmJobRepository } from '../repositories/dvm-job-repository'
 import { EventRepository } from '../repositories/event-repository'
 import { InviteCodeRepository } from '../repositories/invite-code-repository'
 import { Nip05VerificationRepository } from '../repositories/nip05-verification-repository'
 import { ReportRepository } from '../repositories/report-repository'
 import { UserRepository } from '../repositories/user-repository'
+import { createLogger } from './logger-factory'
+import { getCache } from './message-handler-factory'
+import { createWebApp } from './web-app-factory'
 import { webSocketAdapterFactory } from './websocket-adapter-factory'
-import { WebSocketServerAdapter } from '../adapters/web-socket-server-adapter'
+import { wotGraphServiceFactory } from './wot-graph-service-factory'
 
 const logger = createLogger('worker-factory')
 
@@ -29,6 +30,13 @@ export const workerFactory = (): AppWorker => {
   const dvmJobRepository = new DvmJobRepository(dbClient)
   const reportRepository = new ReportRepository(dbClient)
 
+  // Constructs the WoT graph singleton (and starts warming it up, if enabled)
+  // right at worker boot -- before this call, the singleton was only ever
+  // created lazily inside per-event handler wiring, so the very first
+  // EVENT/report needing a WoT distance was also the thing paying for the
+  // cold-start rebuild the warm-up was meant to avoid.
+  wotGraphServiceFactory(getCache(), eventRepository, createSettings)
+
   const settings = createSettings()
 
   const app = createWebApp()

src/factories/wot-graph-service-factory.ts

@@ -13,6 +13,11 @@ export const wotGraphServiceFactory = (
 ): IWotGraphService => {
   if (!instance) {
     instance = new WotGraphService(cache, eventRepository, settings)
+    // Kick off the graph build as soon as the singleton exists (effectively
+    // at worker boot, since this factory runs while wiring up the message
+    // handler/event strategy chains) instead of waiting for the first
+    // report or PoW check to pay for a cold-start rebuild.
+    instance.warmUp()
   }
 
   return instance

src/handlers/event-strategies/report-event-strategy.ts

@@ -1,14 +1,15 @@
-import { createEventCommandResult } from '../../telemetry/event-metrics'
-import { createLogger } from '../../factories/logger-factory'
-import { calculateReportWeight } from '../../utils/report-scoring'
+import { IWebSocketAdapter } from '../../@types/adapters'
 import { Event } from '../../@types/event'
-import { extractReportTargets } from '../../utils/nip56'
-import { IEventRepository, IReportRepository } from '../../@types/repositories'
 import { IEventStrategy } from '../../@types/message-handlers'
-import { IWebSocketAdapter } from '../../@types/adapters'
+import { Report } from '../../@types/report'
+import { IEventRepository, IReportRepository } from '../../@types/repositories'
 import { IWotGraphService } from '../../@types/services'
 import { Settings } from '../../@types/settings'
 import { WebSocketAdapterEvent } from '../../constants/adapter'
+import { createLogger } from '../../factories/logger-factory'
+import { createEventCommandResult } from '../../telemetry/event-metrics'
+import { extractReportTargets } from '../../utils/nip56'
+import { calculateReportWeight } from '../../utils/report-scoring'
 
 const logger = createLogger('report-event-strategy')
 
@@ -51,21 +52,32 @@ export class ReportEventStrategy implements IEventStrategy<Event, Promise<void>>
       const distance = isTrustedModerator ? undefined : await this.wotGraphService.getDistance(event.pubkey)
       const weight = calculateReportWeight(distance, isTrustedModerator)
 
+      const reports: Omit<Report, 'id' | 'createdAt'>[] = []
       for (const target of extractReportTargets(event.tags)) {
-        const hasValidTarget = target.reportedPubkey !== null || target.reportedEventId !== null
+        // A report naming no pubkey/event has nothing to act on -- store the
+        // event itself (already done above) but skip the row instead of
+        // persisting a dead reports entry.
+        if (target.reportedPubkey === null && target.reportedEventId === null) {
+          continue
+        }
 
-        await this.reportRepository.create({
+        reports.push({
           eventId: event.id,
           reporterPubkey: event.pubkey,
           reportedPubkey: target.reportedPubkey,
           reportedEventId: target.reportedEventId,
           reportType: target.reportType,
           weight,
-          // A moderator report with no valid target has nothing to act on --
-          // never mark it actionable regardless of who sent it.
-          actionable: isTrustedModerator && hasValidTarget,
+          actionable: isTrustedModerator,
         })
       }
+
+      if (reports.length) {
+        // One transaction for every target on this event -- a report event
+        // producing more than one row (see extractReportTargets) shouldn't be
+        // able to leave a partial set behind on a mid-batch failure.
+        await this.reportRepository.createMany(reports)
+      }
     } catch (error) {
       // Report scoring/recording is best-effort: the report event itself is
       // already stored and broadcast correctly, so a failure here must not

src/repositories/report-repository.ts

@@ -52,6 +52,25 @@ export class ReportRepository implements IReportRepository {
     return fromDBReport({ ...row, id: inserted.id })
   }
 
+  public async createMany(
+    reports: Omit<Report, 'id' | 'createdAt'>[],
+    client: DatabaseClient = this.dbClient,
+  ): Promise<Report[]> {
+    if (!reports.length) {
+      return []
+    }
+
+    logger('create %d reports in a single transaction', reports.length)
+
+    return client.transaction(async (trx) => {
+      const created: Report[] = []
+      for (const report of reports) {
+        created.push(await this.create(report, trx))
+      }
+      return created
+    })
+  }
+
   public async findByEventId(eventId: EventId, client: DatabaseClient = this.dbClient): Promise<Report[]> {
     logger('find reports for event %s', eventId)
 

src/services/wot-graph-service.ts

@@ -1,10 +1,10 @@
-import { createLogger } from '../factories/logger-factory'
-import { EventKinds, EventTags } from '../constants/base'
 import { ICacheAdapter } from '../@types/adapters'
+import { Pubkey, Tag } from '../@types/base'
 import { IEventRepository } from '../@types/repositories'
 import { IWotGraphService } from '../@types/services'
-import { Pubkey, Tag } from '../@types/base'
 import { Settings } from '../@types/settings'
+import { EventKinds, EventTags } from '../constants/base'
+import { createLogger } from '../factories/logger-factory'
 import { toNostrEvent } from '../utils/event'
 
 const logger = createLogger('wot-graph-service')
@@ -37,6 +37,19 @@ export class WotGraphService implements IWotGraphService {
     return this.ready
   }
 
+  public warmUp(): void {
+    // Do NOT call ensureBuilt() when WoT is disabled: rebuild()'s
+    // disabled-graph branch marks `ready = true` permanently, and if an
+    // operator later hot-enables WoT at runtime, getDistance() would see
+    // `ready` already true and skip rebuilding, leaving the graph
+    // permanently empty until the worker restarts. Only warm up when
+    // there's an actual graph to build.
+    if (!this.settings().wot?.enabled) {
+      return
+    }
+    void this.ensureBuilt()
+  }
+
   public async getDistance(pubkey: Pubkey): Promise<number | undefined> {
     const wot = this.settings().wot
     if (!wot?.enabled || !wot.seedPubkey) {

src/utils/nip56.ts

@@ -1,7 +1,10 @@
 import { EventId, Pubkey, Tag } from '../@types/base'
-import { EventKinds, EventTags } from '../constants/base'
 import { Event } from '../@types/event'
 import { ReportType } from '../@types/report'
+import { EventKinds, EventTags } from '../constants/base'
+import { createLogger } from '../factories/logger-factory'
+
+const logger = createLogger('nip56')
 
 export const isReportEvent = (event: Event): boolean => event.kind === EventKinds.REPORT
 
@@ -25,41 +28,75 @@ export interface ReportTarget {
   reportType: ReportType
 }
 
+const targetTypeOf = (tag: Tag): ReportType => (isValidReportType(tag[2]) ? tag[2] : ReportType.OTHER)
+
+// A single report event can legitimately carry more than one p/e tag (e.g.
+// a moderator batch-reporting a raid of spam accounts in one event); cap how
+// many targets one event can produce so a pathological event can't fan out
+// into an unbounded number of rows, while staying well above any realistic
+// legitimate batch so a genuine report isn't silently truncated.
+const MAX_TARGETS_PER_EVENT = 100
+
 // NIP-56: the report type is the 3rd element of the tag identifying what's
 // being reported. A p tag and an e tag are separate claims (report this
 // pubkey, report this event) that happen to travel in the same event -- when
-// they share a type (the common case: reporting one piece of content and its
-// author for the same reason) they collapse into a single target; when their
-// types disagree, each keeps its own type as its own target rather than one
-// silently overwriting the other's.
+// the *first* p tag and *first* e tag share a type (the common case:
+// reporting one piece of content and its author for the same reason) they
+// collapse into a single target; when their types disagree, each keeps its
+// own type as its own target rather than one silently overwriting the
+// other's. Any additional p/e tags beyond the first pair are independent,
+// single-field targets in their own right.
 export const extractReportTargets = (tags: Tag[]): ReportTarget[] => {
-  const pTag = tags.find((tag) => tag[0] === EventTags.Pubkey && tag.length >= 2 && isValidHex64(tag[1]))
-  const eTag = tags.find((tag) => tag[0] === EventTags.Event && tag.length >= 2 && isValidHex64(tag[1]))
-
-  const pType = isValidReportType(pTag?.[2]) ? pTag![2] : undefined
-  const eType = isValidReportType(eTag?.[2]) ? eTag![2] : undefined
+  const pTags = tags.filter((tag) => tag[0] === EventTags.Pubkey && tag.length >= 2 && isValidHex64(tag[1]))
+  const eTags = tags.filter((tag) => tag[0] === EventTags.Event && tag.length >= 2 && isValidHex64(tag[1]))
 
-  if (!pTag && !eTag) {
+  if (!pTags.length && !eTags.length) {
     return [{ reportedPubkey: null, reportedEventId: null, reportType: ReportType.OTHER }]
   }
 
+  const [pTag, ...extraPTags] = pTags
+  const [eTag, ...extraETags] = eTags
+
+  const targets: ReportTarget[] = []
+
   if (pTag && eTag) {
+    const pType = isValidReportType(pTag[2]) ? pTag[2] : undefined
+    const eType = isValidReportType(eTag[2]) ? eTag[2] : undefined
+
     // Only a genuine disagreement (both sides carry an explicit, different
     // type) splits into two targets; one side simply omitting a type isn't a
     // conflict, so it falls back to whichever side did specify one.
     if (pType !== undefined && eType !== undefined && pType !== eType) {
-      return [
-        { reportedPubkey: null, reportedEventId: eTag[1], reportType: eType },
-        { reportedPubkey: pTag[1], reportedEventId: null, reportType: pType },
-      ]
+      targets.push({ reportedPubkey: null, reportedEventId: eTag[1], reportType: eType })
+      targets.push({ reportedPubkey: pTag[1], reportedEventId: null, reportType: pType })
+    } else {
+      targets.push({
+        reportedPubkey: pTag[1],
+        reportedEventId: eTag[1],
+        reportType: eType ?? pType ?? ReportType.OTHER,
+      })
     }
+  } else if (eTag) {
+    targets.push({ reportedPubkey: null, reportedEventId: eTag[1], reportType: targetTypeOf(eTag) })
+  } else if (pTag) {
+    targets.push({ reportedPubkey: pTag[1], reportedEventId: null, reportType: targetTypeOf(pTag) })
+  }
 
-    return [{ reportedPubkey: pTag[1], reportedEventId: eTag[1], reportType: eType ?? pType ?? ReportType.OTHER }]
+  for (const tag of extraPTags) {
+    targets.push({ reportedPubkey: tag[1], reportedEventId: null, reportType: targetTypeOf(tag) })
+  }
+  for (const tag of extraETags) {
+    targets.push({ reportedPubkey: null, reportedEventId: tag[1], reportType: targetTypeOf(tag) })
   }
 
-  if (eTag) {
-    return [{ reportedPubkey: null, reportedEventId: eTag[1], reportType: eType ?? ReportType.OTHER }]
+  if (targets.length > MAX_TARGETS_PER_EVENT) {
+    logger.error(
+      'report event carries %d targets, exceeding the %d-target cap; dropping the rest',
+      targets.length,
+      MAX_TARGETS_PER_EVENT,
+    )
+    return targets.slice(0, MAX_TARGETS_PER_EVENT)
   }
 
-  return [{ reportedPubkey: pTag![1], reportedEventId: null, reportType: pType ?? ReportType.OTHER }]
+  return targets
 }