feat(notifications): block matcher + gap-limit advancement

matchBlock matches a parsed block's output addresses (incoming) and spent-input
prevout addresses (outgoing) against WatchedAddress, advances the BIP44
gap-limit window (used index i -> derive up to i+gapLimit), and dispatches one
deduped silent push per (subscription, tx, event) for opted-in events only.
Wires it into BlockchainMonitor's new-block (depth 1) and mempool (depth 0,
first-seen) paths behind an injected dispatcher — inert until wired.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: NateIsern

server/lib/blockchain-monitor.ts

@@ -16,13 +16,18 @@ import {
 } from '../../shared/websocket-types'
 import { blockCache } from './cache'
 import { logger } from './logger'
+import { matchBlock, type ParsedBlockTx } from './notifications/matcher'
+import type { PushDispatcher } from './push/types'
 
 /** Default block/mempool poll interval (ms). Overridable via BLOCKCHAIN_POLL_INTERVAL. */
 const DEFAULT_POLL_INTERVAL_MS = 4000
 
 /** Cap on `transaction-confirmed` events emitted per new block. */
 const TRANSACTION_CONFIRMED_CAP = 50
 
+/** Cap on transactions scanned per block/mempool batch for notification matching. */
+const NOTIFICATION_TX_SCAN_CAP = 500
+
 /** Top-N mempool txs included in `mempool-update` (aligned with GET /api/mempool). */
 const MEMPOOL_TX_SUMMARY_LIMIT = 20
 
@@ -80,6 +85,14 @@ export class BlockchainMonitor {
   private lastBroadcastVersion = new Map<NetworkType, string>()
   private lastBroadcastSubversion = new Map<NetworkType, string | undefined>()
   private lastBroadcastProtocol = new Map<NetworkType, number | undefined>()
+  /**
+   * Silent-push dispatcher for background payment notifications. Null (and the
+   * whole notification path inert) until wired at startup with FCM/APNS creds —
+   * Phase-1-safe: no creds → no dispatcher → no matching, no RPC overhead.
+   */
+  private notificationDispatcher: PushDispatcher | null = null
+  /** Mempool txids already notified as first-seen (per network), so a lingering tx notifies once. */
+  private notifiedMempoolTxids = new Map<NetworkType, Set<string>>()
 
   constructor(wsManager: WebSocketManager, config?: Partial<BlockchainMonitorConfig>) {
     this.wsManager = wsManager
@@ -111,6 +124,15 @@ export class BlockchainMonitor {
     logger.debug('[BlockchainMonitor] Initialized with config:', this.config)
   }
 
+  /**
+   * Wire (or clear) the silent-push dispatcher. Called at startup only when
+   * FCM/APNS credentials are configured; while null the notification path is a
+   * complete no-op.
+   */
+  setNotificationDispatcher(dispatch: PushDispatcher | null): void {
+    this.notificationDispatcher = dispatch
+  }
+
   async start(): Promise<void> {
     if (this.isRunning) {
       logger.debug('[BlockchainMonitor] Already running')
@@ -286,6 +308,13 @@ export class BlockchainMonitor {
         }
         this.wsManager.broadcast(confirmedEvent, network)
       }
+
+      // Background payment notifications: match this mined block's txs against
+      // watched addresses and dispatch silent pushes. Isolated below so an RPC or
+      // DB hiccup never disrupts block broadcasting.
+      if (this.notificationDispatcher) {
+        await this.notifyBlockMatches(network, txids)
+      }
     } catch (error) {
       logger.error(
         `[BlockchainMonitor] Error handling new block ${height} on ${network}:`,
@@ -294,6 +323,99 @@ export class BlockchainMonitor {
     }
   }
 
+  /**
+   * Reduce transactions to the address facts the matcher needs. Fetches each
+   * verbose tx (output addresses + resolved input prevout addresses), bounded by
+   * {@link NOTIFICATION_TX_SCAN_CAP}. Best-effort: a tx that fails to load is
+   * skipped rather than aborting the batch.
+   */
+  private async buildParsedTxs(network: NetworkType, txids: string[]): Promise<ParsedBlockTx[]> {
+    const capped = txids.slice(0, NOTIFICATION_TX_SCAN_CAP)
+    if (txids.length > NOTIFICATION_TX_SCAN_CAP) {
+      logger.warn(
+        `[BlockchainMonitor] ${network}: ${txids.length} txs exceed notification scan cap ${NOTIFICATION_TX_SCAN_CAP}; tail unscanned`,
+      )
+    }
+
+    const parsed: ParsedBlockTx[] = []
+    for (const txid of capped) {
+      try {
+        const tx = await blockCache.getTransaction(txid, network, true)
+        if (!tx) {
+          continue
+        }
+        const outputAddresses: string[] = []
+        for (const out of tx.vout ?? []) {
+          for (const address of out.scriptPubKey?.addresses ?? []) {
+            outputAddresses.push(address)
+          }
+        }
+        const inputAddresses: string[] = []
+        for (const input of tx.vin ?? []) {
+          for (const address of input.prevout?.addresses ?? []) {
+            inputAddresses.push(address)
+          }
+        }
+        parsed.push({ txid, outputAddresses, inputAddresses })
+      } catch (error) {
+        logger.debug(`[BlockchainMonitor] notification tx fetch failed for ${txid}:`, error)
+      }
+    }
+    return parsed
+  }
+
+  /** Match a mined block (depth 1) against watched addresses and dispatch pushes. */
+  private async notifyBlockMatches(network: NetworkType, txids: string[]): Promise<void> {
+    const dispatch = this.notificationDispatcher
+    if (!dispatch) {
+      return
+    }
+    try {
+      const parsed = await this.buildParsedTxs(network, txids)
+      if (parsed.length === 0) {
+        return
+      }
+      await matchBlock({ network, depth: 1, txs: parsed }, { dispatch })
+    } catch (error) {
+      logger.error(`[BlockchainMonitor] notification match failed on ${network}:`, error)
+    }
+  }
+
+  /**
+   * Match newly-seen mempool transactions (depth 0) for first-seen
+   * `incoming_pending` alerts. Dedupes against txids already notified while they
+   * linger in the mempool, and prunes that set as txids leave.
+   */
+  private async notifyMempoolMatches(network: NetworkType, rawMempool: string[]): Promise<void> {
+    const dispatch = this.notificationDispatcher
+    if (!dispatch) {
+      return
+    }
+    try {
+      const notified = this.notifiedMempoolTxids.get(network) ?? new Set<string>()
+      const fresh = rawMempool.filter((txid) => !notified.has(txid))
+      if (fresh.length > 0) {
+        const parsed = await this.buildParsedTxs(network, fresh)
+        if (parsed.length > 0) {
+          await matchBlock({ network, depth: 0, txs: parsed }, { dispatch })
+        }
+        for (const txid of fresh) {
+          notified.add(txid)
+        }
+      }
+      // Prune txids that have left the mempool so the set stays bounded.
+      const inMempool = new Set(rawMempool)
+      for (const txid of notified) {
+        if (!inMempool.has(txid)) {
+          notified.delete(txid)
+        }
+      }
+      this.notifiedMempoolTxids.set(network, notified)
+    } catch (error) {
+      logger.error(`[BlockchainMonitor] mempool notification match failed on ${network}:`, error)
+    }
+  }
+
   /**
    * Top-N mempool summary — same fields and limit as GET /api/mempool.
    */
@@ -363,6 +485,14 @@ export class BlockchainMonitor {
           },
         }
         this.wsManager.broadcast(mempoolEvent, network)
+
+        // First-seen payment notifications from the mempool (depth 0).
+        if (this.notificationDispatcher) {
+          const rawMempool = await rpcWithNetwork<string[]>('getrawmempool', [], network).catch(
+            () => [] as string[],
+          )
+          await this.notifyMempoolMatches(network, rawMempool)
+        }
       }
     } catch (error) {
       logger.error(`[BlockchainMonitor] Error polling mempool for ${network}:`, error)

server/lib/notifications/matcher.test.ts

@@ -0,0 +1,185 @@
+import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'
+import { setupMemoryMongo, teardownMemoryMongo, clearCollections } from './test-support'
+import { registerSubscription } from './subscriptions'
+import { deriveWindow } from './derive'
+import { matchBlock, type ParsedBlock } from './matcher'
+import type { PushTarget, PushMessage } from '../push/types'
+import WatchedAddress from '../db/models/WatchedAddress'
+import NotificationSubscription from '../db/models/NotificationSubscription'
+
+const XPUB =
+  'ToEA6m5JcxaE72JyzLAvHoonWB3GgAxzt5jADQC9Pc43q4V3omj94DdKe98KkRgY2nqwbNHxWCceM3o62VhFVe7HCnAW7yBGEvfbfwrYsK3N7t7'
+const GAP_LIMIT = 5
+
+/** Address at a receive/change index for the fixed test xpub. */
+function addr(chain: 0 | 1, index: number): string {
+  return deriveWindow(XPUB, chain, index, 1, 'mainnet')[0].address
+}
+
+async function seedSubscription(
+  events = ['incoming_pending', 'incoming_confirmed', 'outgoing_confirmed'],
+): Promise<string> {
+  const { subscriptionId } = await registerSubscription({
+    xpub: XPUB,
+    scriptType: 'p2pkh',
+    gapLimit: GAP_LIMIT,
+    network: 'mainnet',
+    deviceToken: 'device-token-abc',
+    platform: 'android',
+    confirmations: 1,
+    events: events as never,
+  })
+  return subscriptionId
+}
+
+/** Collect dispatcher calls. */
+function recordingDispatcher() {
+  const calls: Array<{ target: PushTarget; message: PushMessage }> = []
+  return {
+    calls,
+    dispatch: async (target: PushTarget, message: PushMessage) => {
+      calls.push({ target, message })
+    },
+  }
+}
+
+describe('matchBlock', () => {
+  beforeAll(async () => {
+    await setupMemoryMongo()
+  }, 60000)
+  afterAll(async () => {
+    await teardownMemoryMongo()
+  })
+  beforeEach(async () => {
+    await clearCollections()
+  })
+
+  it('dispatches incoming_pending for a first-seen output and advances the gap window', async () => {
+    const subscriptionId = await seedSubscription()
+    const recorder = recordingDispatcher()
+
+    const block: ParsedBlock = {
+      network: 'mainnet',
+      depth: 0, // mempool / first-seen
+      txs: [
+        {
+          txid: 'a'.repeat(64),
+          outputAddresses: [addr(0, 3)],
+          inputAddresses: [],
+        },
+      ],
+    }
+    await matchBlock(block, { dispatch: recorder.dispatch })
+
+    expect(recorder.calls).toHaveLength(1)
+    expect(recorder.calls[0].target.subscriptionId).toBe(subscriptionId)
+    expect(recorder.calls[0].message).toEqual({ txid: 'a'.repeat(64), event: 'incoming_pending' })
+
+    // Using receive index 3 extends the watched window to 3 + gapLimit.
+    const sub = await NotificationSubscription.findOne({ subscriptionId })
+    expect(sub?.derivedTo.receive).toBe(3 + GAP_LIMIT)
+    const receiveRows = await WatchedAddress.find({ subscriptionId, chain: 0 })
+    expect(receiveRows).toHaveLength(3 + GAP_LIMIT)
+    // The new frontier index (3+gapLimit-1) is now watched.
+    expect(await WatchedAddress.countDocuments({ subscriptionId, chain: 0, index: 3 + GAP_LIMIT - 1 })).toBe(1)
+  })
+
+  it('dispatches incoming_confirmed when a mined block reaches the confirmation depth', async () => {
+    const subscriptionId = await seedSubscription()
+    const recorder = recordingDispatcher()
+
+    const block: ParsedBlock = {
+      network: 'mainnet',
+      depth: 1, // confirmations default is 1
+      txs: [{ txid: 'b'.repeat(64), outputAddresses: [addr(0, 0)], inputAddresses: [] }],
+    }
+    await matchBlock(block, { dispatch: recorder.dispatch })
+
+    expect(recorder.calls).toHaveLength(1)
+    expect(recorder.calls[0].message).toEqual({ txid: 'b'.repeat(64), event: 'incoming_confirmed' })
+    expect(recorder.calls[0].target.subscriptionId).toBe(subscriptionId)
+  })
+
+  it('dispatches outgoing_confirmed when a mined tx spends a watched output', async () => {
+    await seedSubscription()
+    const recorder = recordingDispatcher()
+
+    const block: ParsedBlock = {
+      network: 'mainnet',
+      depth: 1,
+      txs: [
+        {
+          txid: 'c'.repeat(64),
+          outputAddresses: ['FUnrelatedAddressNotWatched1111111'],
+          inputAddresses: [addr(0, 0)], // spending a previously watched output
+        },
+      ],
+    }
+    await matchBlock(block, { dispatch: recorder.dispatch })
+
+    expect(recorder.calls).toHaveLength(1)
+    expect(recorder.calls[0].message.event).toBe('outgoing_confirmed')
+  })
+
+  it('sends only one push when a tx pays two watched addresses of the same subscription', async () => {
+    await seedSubscription()
+    const recorder = recordingDispatcher()
+
+    const block: ParsedBlock = {
+      network: 'mainnet',
+      depth: 0,
+      txs: [
+        {
+          txid: 'd'.repeat(64),
+          outputAddresses: [addr(0, 0), addr(0, 1)],
+          inputAddresses: [],
+        },
+      ],
+    }
+    await matchBlock(block, { dispatch: recorder.dispatch })
+
+    expect(recorder.calls).toHaveLength(1)
+  })
+
+  it('does not dispatch an event the subscription did not opt into', async () => {
+    await seedSubscription(['incoming_pending', 'incoming_confirmed']) // no outgoing
+    const recorder = recordingDispatcher()
+
+    const block: ParsedBlock = {
+      network: 'mainnet',
+      depth: 1,
+      txs: [{ txid: 'e'.repeat(64), outputAddresses: [], inputAddresses: [addr(0, 0)] }],
+    }
+    await matchBlock(block, { dispatch: recorder.dispatch })
+
+    expect(recorder.calls).toHaveLength(0)
+  })
+
+  it('does not dispatch for a block with no watched addresses', async () => {
+    await seedSubscription()
+    const recorder = recordingDispatcher()
+
+    const block: ParsedBlock = {
+      network: 'mainnet',
+      depth: 1,
+      txs: [{ txid: 'f'.repeat(64), outputAddresses: ['FSomeoneElse'], inputAddresses: [] }],
+    }
+    await matchBlock(block, { dispatch: recorder.dispatch })
+
+    expect(recorder.calls).toHaveLength(0)
+  })
+
+  it('does not match a subscription on a different network', async () => {
+    await seedSubscription() // mainnet
+    const recorder = recordingDispatcher()
+
+    const block: ParsedBlock = {
+      network: 'testnet',
+      depth: 0,
+      txs: [{ txid: '0'.repeat(64), outputAddresses: [addr(0, 0)], inputAddresses: [] }],
+    }
+    await matchBlock(block, { dispatch: recorder.dispatch })
+
+    expect(recorder.calls).toHaveLength(0)
+  })
+})