Re-validate pending sol_purchases when blocknumber catches up (#818)

## Summary

- Adds a trigger on `tracks`/`playlists` that emits `NOTIFY
'pending_purchase_revalidation'` when `blocknumber` advances past any
pending `sol_purchases` row's `valid_after_blocknumber`. Cheap EXISTS
guard so the vast majority of track/playlist updates skip the notify
entirely.
- Adds a Go listener (`PurchaseRevalidator`) in the Solana indexer that
consumes notifications and re-runs the existing `validatePurchase` for
affected pending rows. A 5-minute sweep + startup sweep covers cases
where NOTIFY drops (no connected listener) or pending rows that predate
the trigger.
- Validation logic stays in `validatePurchase` as the single source of
truth — the trigger only signals which `content_id` changed. No SQL
port, no parity drift with config-driven values like `NetworkTakeRate`.

### Why

A `sol_purchases` row is inserted with `is_valid = NULL` ("pending")
when its `valid_after_blocknumber` hasn't been indexed by the local node
yet (`MAX(blocks.number) < valid_after_blocknumber` at insert time).
Before this PR there was no mechanism to flip pending rows to a final
verdict, so they stayed `NULL` indefinitely even after the indexer
caught up.

### Notes for reviewers

- `sql/01_schema.sql` diff is large because the dump hadn't been
regenerated since #815 merged — it picks up that PR's
`handle_sol_purchase` function and a few other already-merged trigger
updates alongside this PR's `notify_pending_purchase_revalidation`.
- Bumped `pgxpool.MaxConns` by 1 to budget for the revalidator's
persistent LISTEN connection (in addition to the existing artist_coins
listener).
- Lifecycle is purely ctx-driven — no explicit `Stop()`. Cancelling the
parent ctx unblocks `WaitForNotification`, the deferred `conn.Release()`
returns the LISTEN connection to the pool, both goroutines exit.

## Test plan

- [x] Trigger function applied via regenerated schema dump (`make
test-schema`)
- [x] `TestParseRevalidationPayload` — unit coverage for payload parsing
- [x] `TestRevalidatorRevalidateContent` — direct call: pending row +
matching `sol_payments` flips to `true`
- [x] `TestRevalidatorEndToEnd` — full trigger → NOTIFY → LISTEN →
revalidate chain via `UPDATE tracks SET blocknumber`
- [x] Existing `TestPurchaseValidation` still passes
- [ ] Deploy to stage and watch a pending row resolve in the wild

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Author: rickyrombo

ddl/functions/notify_pending_purchase_revalidation.sql

@@ -0,0 +1,70 @@
+-- Notifies Go to re-run validatePurchase for sol_purchases rows that became
+-- eligible because a tracks/playlists row's blocknumber advanced past their
+-- valid_after_blocknumber. The trigger does no validation itself — it just
+-- signals which content_id changed; the Go indexer LISTENs on
+-- 'pending_purchase_revalidation' and re-runs the existing validator so the
+-- math stays in one place. A periodic sweep in Go catches anything missed
+-- (NOTIFY is dropped if no listener is connected at the moment).
+create or replace function notify_pending_purchase_revalidation() returns trigger as $$
+declare
+  v_content_type text;
+  v_content_id int;
+  v_blocknumber int;
+begin
+  if tg_table_name = 'tracks' then
+    v_content_type := 'track';
+    v_content_id := new.track_id;
+  elsif tg_table_name = 'playlists' then
+    v_content_type := 'album';
+    v_content_id := new.playlist_id;
+  else
+    return null;
+  end if;
+
+  v_blocknumber := new.blocknumber;
+  if v_blocknumber is null then
+    return null;
+  end if;
+
+  -- Cheap EXISTS guard: tracks/playlists update churn dwarfs the pending
+  -- purchase set, so it's almost always a no-op. Uses sol_purchases_valid_idx
+  -- + sol_purchases_content_idx.
+  if exists (
+    select 1 from sol_purchases sp
+    where sp.content_id = v_content_id
+      and sp.content_type = v_content_type
+      and sp.is_valid is null
+      and sp.valid_after_blocknumber <= v_blocknumber
+  ) then
+    perform pg_notify(
+      'pending_purchase_revalidation',
+      v_content_type || ':' || v_content_id::text
+    );
+  end if;
+
+  return null;
+exception
+  when others then
+    -- Never let a notify failure break a tracks/playlists write. The sweep
+    -- will catch any pending rows that don't get notified.
+    raise warning 'An error occurred in %: %', tg_name, sqlerrm;
+    return null;
+end;
+$$ language plpgsql;
+
+
+do $$ begin
+  create trigger on_track_notify_pending_purchase_revalidation
+  after insert or update of blocknumber on tracks
+  for each row execute procedure notify_pending_purchase_revalidation();
+exception
+  when others then null;
+end $$;
+
+do $$ begin
+  create trigger on_playlist_notify_pending_purchase_revalidation
+  after insert or update of blocknumber on playlists
+  for each row execute procedure notify_pending_purchase_revalidation();
+exception
+  when others then null;
+end $$;

solana/indexer/program/indexer.go

@@ -26,6 +26,7 @@ type Indexer struct {
 	rpcClient        common.RpcClient
 	config           config.Config
 	transactionCache *otter.Cache[solana.Signature, *rpc.GetTransactionResult]
+	revalidator      *Revalidator
 	logger           *zap.Logger
 }
 
@@ -37,17 +38,21 @@ func New(
 	transactionCache *otter.Cache[solana.Signature, *rpc.GetTransactionResult],
 	logger *zap.Logger,
 ) *Indexer {
+	namedLogger := logger.Named(NAME)
 	return &Indexer{
 		pool:             pool,
 		grpcConfig:       grpcConfig,
 		rpcClient:        rpcClient,
 		config:           config,
 		transactionCache: transactionCache,
-		logger:           logger.Named(NAME),
+		revalidator:      NewRevalidator(pool, config, namedLogger),
+		logger:           namedLogger,
 	}
 }
 
 func (d *Indexer) Start(ctx context.Context) {
+	d.revalidator.Start(ctx)
+
 	client, err := d.subscribe(ctx)
 	if err != nil {
 		d.logger.Fatal("failed to start subscription", zap.Error(err))