From bae30d592c38d83f3b25da754e3fe27dfffbcdca Mon Sep 17 00:00:00 2001
From: macjayz <macjayzz1@gmail.com>
Date: Sun, 1 Nov 2026 23:03:47 +0100
Subject: [PATCH] Keep per-event threat context queries to their time windows

The worker runs these for every evm_rpc event. Three of them filtered by time only inside
count(*) filter, so each read the endpoint's or provider's whole history, and the endpoint
match used is not distinct from, which no index can serve. On the study database (650k rows)
the worker fell behind the stream and stopped saving events.

Bound the queries to their windows, compare a known endpoint hash with =, and index provider
and wallet lookups. Results are unchanged.
---
 apps/worker/src/db.ts                         | 23 +++++++++++++++----
 .../migrations/004_threat_context_indexes.sql | 14 +++++++++++
 2 files changed, 32 insertions(+), 5 deletions(-)
 create mode 100644 infra/migrations/004_threat_context_indexes.sql

diff --git a/apps/worker/src/db.ts b/apps/worker/src/db.ts
index c05827b..da0602f 100644
--- a/apps/worker/src/db.ts
+++ b/apps/worker/src/db.ts
@@ -34,6 +34,16 @@ export function createPool(config: WorkerConfig) {
   return pool;
 }
 
+/**
+ * Matches the endpoint hash, null included. Postgres cannot use an index for `is not distinct
+ * from`, so a known hash is compared with `=`. The parameter stays referenced either way.
+ */
+function endpointMatch(endpointHash: string | null | undefined, param: number) {
+  return endpointHash == null
+    ? `evm_endpoint_hash is null and $${param}::text is null`
+    : `evm_endpoint_hash = $${param}`;
+}
+
 export async function countRecentIpRequests(pool: pg.Pool, projectId: string, ip?: string) {
   if (!ip) return 0;
   const result = await pool.query(
@@ -87,6 +97,7 @@ export async function getWalletTransactionStats(
     from api_events
     where project_id = $1 and kind = 'evm_rpc' and wallet_address = $2
       and evm_rpc_method ~* 'sendrawtransaction|sendtransaction'
+      and timestamp > now() - interval '25 hours'
     `,
     [projectId, walletAddress]
   );
@@ -114,6 +125,7 @@ export async function getProviderRpcStats(pool: pg.Pool, projectId: string, prov
       )::int as recent_failures
     from api_events
     where project_id = $1 and kind = 'evm_rpc' and evm_provider = $2
+      and timestamp > now() - interval '24 hours'
     `,
     [projectId, provider]
   );
@@ -158,7 +170,7 @@ export async function getStaleHeadContext(
       select timestamp, evm_block_number
       from api_events
       where project_id = $1 and kind = 'evm_rpc' and evm_chain_id = $2
-        and evm_endpoint_hash is not distinct from $3
+        and ${endpointMatch(endpointHash, 3)}
         and evm_block_number is not null
         and (evm_block_tag is null or evm_block_tag = 'latest')
         and evm_shadow_of_trace_id is null
@@ -233,7 +245,7 @@ export async function getSilentFailureContext(
       select exists (
         select 1 from api_events
         where project_id = $1 and kind = 'evm_rpc' and evm_rpc_method = $2
-          and evm_endpoint_hash is not distinct from $3
+          and ${endpointMatch(endpointHash, 3)}
           and request_body -> 'params' ->> 0 = $4
           and evm_result_shape is not null and evm_result_shape <> 'null'
           and timestamp > now() - interval '24 hours'
@@ -252,7 +264,7 @@ export async function getSilentFailureContext(
       select exists (
         select 1 from api_events
         where project_id = $1 and kind = 'evm_rpc' and evm_rpc_method = $2
-          and evm_endpoint_hash is not distinct from $3
+          and ${endpointMatch(endpointHash, 3)}
           and request_body -> 'params' -> 0 = $4::jsonb
           and evm_result_shape = 'value'
           and (evm_block_number is null or $5::bigint is null or evm_block_number < $5::bigint)
@@ -270,7 +282,7 @@ export async function getSilentFailureContext(
       select evm_result_count as max_count, count(*)::int as hits
       from api_events
       where project_id = $1 and kind = 'evm_rpc' and evm_rpc_method = $2
-        and evm_endpoint_hash is not distinct from $3
+        and ${endpointMatch(endpointHash, 3)}
         and evm_result_count is not null
         and timestamp > now() - interval '24 hours'
       group by evm_result_count
@@ -480,7 +492,8 @@ export async function getThrottlingContext(
       )::int as baseline_degraded
     from api_events
     where project_id = $1 and kind = 'evm_rpc'
-      and evm_endpoint_hash is not distinct from $2
+      and ${endpointMatch(endpointHash, 2)}
+      and timestamp >= now() - interval '24 hours'
     `,
     [projectId, endpointHash ?? null]
   );
diff --git a/infra/migrations/004_threat_context_indexes.sql b/infra/migrations/004_threat_context_indexes.sql
new file mode 100644
index 0000000..d9f299d
--- /dev/null
+++ b/infra/migrations/004_threat_context_indexes.sql
@@ -0,0 +1,14 @@
+-- Indexes for the per-event threat context queries.
+--
+-- The worker runs these for every evm_rpc event, so each must read only its time window, not
+-- the provider's or wallet's whole history.
+
+-- Provider latency and failure rate (provider_degradation, D5).
+create index if not exists api_events_evm_provider_idx
+  on api_events(project_id, evm_provider, timestamp desc)
+  where kind = 'evm_rpc';
+
+-- Wallet transaction rate (wallet_tx_spike).
+create index if not exists api_events_evm_wallet_idx
+  on api_events(project_id, wallet_address, timestamp desc)
+  where kind = 'evm_rpc' and wallet_address is not null;