Switch challenge reads to sol_reward_disbursements

api/dbv1/queries/get_undisbursed_challenges.sql

@@ -7,7 +7,7 @@ SELECT
     user_challenges.amount
 FROM user_challenges
 JOIN users ON users.user_id = user_challenges.user_id
-LEFT JOIN challenge_disbursements
+LEFT JOIN v_challenge_disbursements AS challenge_disbursements
     ON challenge_disbursements.challenge_id = user_challenges.challenge_id
     AND challenge_disbursements.specifier = user_challenges.specifier
 WHERE

api/v1_challenges_disbursements.go

@@ -79,7 +79,7 @@ func (app *ApiServer) v1ChallengesDisbursements(c *fiber.Ctx) error {
 		cd.created_at,
 		cd.signature,
 		cd.slot
-	FROM challenge_disbursements cd
+	FROM v_challenge_disbursements cd
 	` + whereClause + `
 	ORDER BY ` + sortMethod + ` ` + sortDir + `, cd.user_id ASC
 	LIMIT @limit

api/v1_challenges_disbursements_test.go

@@ -13,33 +13,41 @@ func TestGetChallengeDisbursements(t *testing.T) {
 	app := emptyTestApp(t)
 
 	fixtures := database.FixtureMap{
-		"challenge_disbursements": {
+		"users": {
+			{"user_id": 1, "handle": "user1", "handle_lc": "user1", "wallet": "0xwallet1"},
+			{"user_id": 2, "handle": "user2", "handle_lc": "user2", "wallet": "0xwallet2"},
+			{"user_id": 3, "handle": "user3", "handle_lc": "user3", "wallet": "0xwallet3"},
+		},
+		"sol_reward_disbursements": {
 			{
-				"challenge_id": "e",
-				"user_id":      1,
-				"specifier":    "def",
-				"signature":    "sig123",
-				"slot":         102,
-				"amount":       "100",
-				"created_at":   time.Now().Add(-time.Minute * 3),
+				"challenge_id":          "e",
+				"specifier":             "def",
+				"signature":             "sig123",
+				"instruction_index":     0,
+				"slot":                  102,
+				"amount":                100,
+				"recipient_eth_address": "0xwallet1",
+				"created_at":            time.Now().Add(-time.Minute * 3),
 			},
 			{
-				"challenge_id": "f",
-				"user_id":      3,
-				"specifier":    "jkl:3",
-				"signature":    "sig789",
-				"slot":         101,
-				"amount":       "200",
-				"created_at":   time.Now().Add(-time.Minute * 2),
+				"challenge_id":          "f",
+				"specifier":             "jkl:3",
+				"signature":             "sig789",
+				"instruction_index":     0,
+				"slot":                  101,
+				"amount":                200,
+				"recipient_eth_address": "0xwallet3",
+				"created_at":            time.Now().Add(-time.Minute * 2),
 			},
 			{
-				"challenge_id": "f",
-				"user_id":      2,
-				"specifier":    "jkl:2",
-				"signature":    "sig456",
-				"slot":         103,
-				"amount":       "200",
-				"created_at":   time.Now().Add(-time.Minute * 1),
+				"challenge_id":          "f",
+				"specifier":             "jkl:2",
+				"signature":             "sig456",
+				"instruction_index":     0,
+				"slot":                  103,
+				"amount":                200,
+				"recipient_eth_address": "0xwallet2",
+				"created_at":            time.Now().Add(-time.Minute * 1),
 			},
 		},
 	}

api/v1_challenges_info.go

@@ -60,7 +60,7 @@ func (app *ApiServer) v1ChallengesInfo(c *fiber.Ctx) error {
 			WHEN c.weekly_pool IS NULL THEN NULL
 			ELSE c.weekly_pool - COALESCE(
 				(SELECT SUM(cd.amount::bigint) / 100000000
-				 FROM challenge_disbursements cd
+				 FROM v_challenge_disbursements cd
 				 WHERE cd.challenge_id = c.id
 				   AND cd.created_at > @weeklyPoolWindowStart),
 				0

api/v1_challenges_info_test.go

@@ -26,15 +26,19 @@ func TestV1ChallengesInfo(t *testing.T) {
 				"cooldown_days":  7,
 			},
 		},
-		"challenge_disbursements": {
+		"users": {
+			{"user_id": 1, "handle": "user1", "handle_lc": "user1", "wallet": "0xwallet1"},
+		},
+		"sol_reward_disbursements": {
 			{
-				"challenge_id": "challenge-aggregate",
-				"user_id":      1,
-				"specifier":    "spec-a",
-				"signature":    "sig-a",
-				"slot":         1,
-				"amount":       "200000000",
-				"created_at":   now,
+				"challenge_id":          "challenge-aggregate",
+				"specifier":             "spec-a",
+				"signature":             "sig-a",
+				"instruction_index":     0,
+				"slot":                  1,
+				"amount":                200000000,
+				"recipient_eth_address": "0xwallet1",
+				"created_at":            now,
 			},
 		},
 	}

api/v1_challenges_undisbursed.go

@@ -62,7 +62,7 @@ func (app *ApiServer) v1ChallengesUndisbursed(c *fiber.Ctx) error {
 	FROM user_challenges
 	JOIN challenges ON challenges.id = user_challenges.challenge_id
 	JOIN users ON users.user_id = user_challenges.user_id
-	LEFT JOIN challenge_disbursements ON
+	LEFT JOIN v_challenge_disbursements AS challenge_disbursements ON
 		challenge_disbursements.challenge_id = user_challenges.challenge_id
 		AND challenge_disbursements.specifier = user_challenges.specifier
 	WHERE challenge_disbursements.challenge_id IS NULL

api/v1_coins_post_redeem.go

@@ -163,7 +163,7 @@ func (app *ApiServer) v1CoinsPostRedeem(c *fiber.Ctx) error {
 		redeemCode = coinTicker
 		// Check for challenge disbursement for the given code/userId
 		var count int
-		err := app.writePool.QueryRow(c.Context(), `SELECT count(*) FROM challenge_disbursements WHERE challenge_id = @code AND specifier = @specifier LIMIT 1;`, pgx.NamedArgs{
+		err := app.writePool.QueryRow(c.Context(), `SELECT count(*) FROM v_challenge_disbursements WHERE challenge_id = @code AND specifier = @specifier LIMIT 1;`, pgx.NamedArgs{
 			"code":      redeemCode,
 			"specifier": specifier,
 		}).Scan(&count)

api/v1_users_challenges.go

@@ -26,7 +26,7 @@ func (app *ApiServer) v1UsersChallenges(c *fiber.Ctx) error {
 
 	-- Pre-filter to their disbursements
 	challenge_disbursements_filtered AS (
-		SELECT * FROM challenge_disbursements JOIN user_row USING (user_id)
+		SELECT * FROM v_challenge_disbursements JOIN user_row USING (user_id)
 	),
 
 	-- Start with the list of all active challenges, and then

database/seed.go

@@ -520,6 +520,17 @@ var (
 			"ethereum_address":  nil,
 			"account":           nil,
 		},
+		"sol_reward_disbursements": {
+			"signature":             nil,
+			"instruction_index":     0,
+			"amount":                nil,
+			"slot":                  1,
+			"user_bank":             "user-bank-placeholder",
+			"challenge_id":          nil,
+			"specifier":             nil,
+			"recipient_eth_address": nil,
+			"created_at":            time.Now(),
+		},
 		"sol_token_account_balance_changes": {
 			"account":         nil,
 			"owner":           "owner-acc",

ddl/functions/handle_challenge_disbursements.sql

@@ -56,3 +56,85 @@ do $$ begin
 exception
   when others then null;
 end $$;
+
+
+-- Mirror of handle_challenge_disbursement for the new indexer's table.
+-- Resolves user_id from user_bank via sol_claimable_accounts -> users.
+-- Notification shape and dedupe group_id match the legacy trigger so the two
+-- can dual-fire during cutover (the second insert is a no-op via on conflict).
+-- Also pg_notify's so the Python ChallengeEventBus can subscribe once the
+-- legacy index_rewards_manager Celery task is decommissioned.
+create or replace function handle_sol_reward_disbursement() returns trigger as $$
+declare
+  resolved_user_id integer;
+  existing_notification integer;
+  reward_code_exists boolean;
+begin
+  select users.user_id
+    into resolved_user_id
+    from users
+   where users.wallet = new.recipient_eth_address
+     and users.is_current = true
+   limit 1;
+
+  if resolved_user_id is null then
+    return null;
+  end if;
+
+  select exists(select 1 from reward_codes where code = new.challenge_id) into reward_code_exists;
+
+  if not reward_code_exists then
+    select id into existing_notification
+      from notification
+     where type = 'challenge_reward'
+       and resolved_user_id = any(user_ids)
+       and timestamp >= (new.created_at - interval '1 hour')
+     limit 1;
+
+    if existing_notification is null then
+      insert into notification
+        (slot, user_ids, timestamp, type, group_id, specifier, data)
+      values
+        (
+          new.slot,
+          ARRAY [resolved_user_id],
+          new.created_at,
+          'challenge_reward',
+          'challenge_reward:' || resolved_user_id || ':challenge:' || new.challenge_id || ':specifier:' || new.specifier,
+          resolved_user_id,
+          json_build_object('specifier', new.specifier, 'challenge_id', new.challenge_id, 'amount', new.amount)
+        )
+        on conflict do nothing;
+    end if;
+  end if;
+
+  perform pg_notify(
+    'challenge_disbursed',
+    json_build_object(
+      'user_id', resolved_user_id,
+      'challenge_id', new.challenge_id,
+      'specifier', new.specifier,
+      'amount', new.amount,
+      'signature', new.signature,
+      'slot', new.slot
+    )::text
+  );
+
+  return null;
+
+exception
+  when others then
+    raise warning 'An error occurred in %: %', tg_name, sqlerrm;
+    return null;
+
+end;
+$$ language plpgsql;
+
+
+do $$ begin
+  create trigger on_sol_reward_disbursement
+    after insert on sol_reward_disbursements
+    for each row execute procedure handle_sol_reward_disbursement();
+exception
+  when others then null;
+end $$;

ddl/migrations/0198_sol_reward_disbursements_created_at.sql

@@ -0,0 +1,25 @@
+BEGIN;
+
+-- Parity with the legacy challenge_disbursements.created_at column. The Go indexer
+-- writes rows close to on-chain time, so DEFAULT NOW() is acceptable for new rows;
+-- backfilled rows are corrected from the legacy table below.
+ALTER TABLE sol_reward_disbursements
+    ADD COLUMN IF NOT EXISTS created_at TIMESTAMP DEFAULT NOW();
+
+UPDATE sol_reward_disbursements rd
+   SET created_at = cd.created_at
+  FROM challenge_disbursements cd
+ WHERE cd.signature = rd.signature
+   AND cd.created_at IS NOT NULL
+   AND (rd.created_at IS NULL OR rd.created_at > cd.created_at);
+
+-- Restore the indexes the legacy challenge_disbursements table had so the API
+-- routes that filter/sort by user (via recipient_eth_address) and created_at
+-- don't degenerate to seq scans through v_challenge_disbursements.
+CREATE INDEX IF NOT EXISTS sol_reward_disbursements_recipient_eth_address_idx
+    ON sol_reward_disbursements (recipient_eth_address);
+
+CREATE INDEX IF NOT EXISTS sol_reward_disbursements_created_at_idx
+    ON sol_reward_disbursements (created_at);
+
+COMMIT;

ddl/views/v_challenge_disbursements.sql

@@ -0,0 +1,16 @@
+DROP VIEW IF EXISTS v_challenge_disbursements;
+CREATE VIEW v_challenge_disbursements AS
+    SELECT
+        rd.challenge_id,
+        rd.specifier,
+        rd.amount::text AS amount,
+        rd.signature,
+        rd.slot,
+        rd.created_at,
+        users.user_id
+    FROM sol_reward_disbursements rd
+    JOIN users
+        ON users.wallet = rd.recipient_eth_address
+       AND users.is_current = TRUE;
+
+COMMENT ON VIEW v_challenge_disbursements IS 'Compatibility view that exposes sol_reward_disbursements in the column shape the API routes used to read from challenge_disbursements. Resolves user_id via the indexer-populated recipient_eth_address (see migration 0172).';

sql/01_schema.sql

@@ -8413,7 +8413,8 @@ CREATE TABLE public.sol_reward_disbursements (
     user_bank character varying NOT NULL,
     challenge_id character varying NOT NULL,
     specifier character varying NOT NULL,
-    recipient_eth_address text
+    recipient_eth_address text,
+    created_at timestamp without time zone DEFAULT now()
 );
 
 
@@ -12872,6 +12873,22 @@ ALTER TABLE ONLY public.users
     ADD CONSTRAINT users_blocknumber_fkey FOREIGN KEY (blocknumber) REFERENCES public.blocks(number) ON DELETE CASCADE;
 
 
+--
+-- Name: v_challenge_disbursements; Type: VIEW; Schema: public; Owner: -
+--
+
+CREATE VIEW public.v_challenge_disbursements AS
+ SELECT rd.challenge_id,
+    rd.specifier,
+    (rd.amount)::text AS amount,
+    rd.signature,
+    rd.slot,
+    rd.created_at,
+    users.user_id
+   FROM (public.sol_reward_disbursements rd
+     JOIN public.users ON (((users.wallet = rd.recipient_eth_address) AND (users.is_current = true))));
+
+
 --
 -- PostgreSQL database dump complete
 --