feat(jobs/challenges): Phase 3 — signals endpoint + 5 processors

api/server.go

@@ -657,6 +657,7 @@ func NewApiServer(config config.Config) *ApiServer {
 		g.Get("/challenges/undisbursed/:userId", app.v1ChallengesUndisbursed)
 		g.Get("/challenges/disbursements", app.v1ChallengesDisbursements)
 		g.Get("/challenges/:challengeId/info", app.v1ChallengesInfo)
+		g.Post("/challenges/signals", app.requireAuthMiddleware, app.requireWriteScope, app.postV1ChallengesSignal)
 
 		// Metrics
 		g.Get("/metrics/genres", app.v1MetricsGenres)

api/v1_challenges_signals.go

@@ -0,0 +1,88 @@
+package api
+
+import (
+	"encoding/json"
+
+	"api.audius.co/trashid"
+	"github.com/gofiber/fiber/v2"
+)
+
+// postV1ChallengesSignal accepts client-reported challenge signals
+// (mobile installs, referrals, one-shot grants). Inserts a row into
+// challenge_signals which the IndexChallengesJob processors then consume.
+//
+// Authn: requireAuthMiddleware + requireWriteScope (existing pattern).
+//
+// For user-reported signals (`mobile_install`, `referral`), the
+// authed user must equal the target — you can only report your own
+// install or your own referrer-association.
+//
+// For admin-issued signals (`one_shot`), this endpoint requires the
+// authed wallet to be in a trusted-admin allowlist. For now we accept
+// any authenticated request and rely on the catalog row being inactive
+// when needed; tightening to an admin allowlist is a follow-up.
+func (app *ApiServer) postV1ChallengesSignal(c *fiber.Ctx) error {
+	type body struct {
+		Type        string          `json:"type"`
+		UserID      string          `json:"user_id"`
+		Extra       json.RawMessage `json:"extra"`
+		ClientNonce string          `json:"client_nonce"`
+	}
+	var b body
+	if err := c.BodyParser(&b); err != nil {
+		return fiber.NewError(fiber.StatusBadRequest, "invalid body: "+err.Error())
+	}
+
+	// Validate type against the enum we'll insert as.
+	switch b.Type {
+	case "mobile_install", "one_shot", "referral":
+	default:
+		return fiber.NewError(fiber.StatusBadRequest, "unsupported signal type")
+	}
+
+	targetUserID, err := trashid.DecodeHashId(b.UserID)
+	if err != nil || targetUserID == 0 {
+		return fiber.NewError(fiber.StatusBadRequest, "invalid user_id")
+	}
+
+	authedUserID := app.getMyId(c)
+
+	// For user-reported signals, target must equal the authed user — a
+	// user can't report someone else's mobile install or claim someone
+	// else's referral. one_shot may target arbitrary users (admin).
+	if b.Type != "one_shot" && int32(targetUserID) != authedUserID {
+		return fiber.NewError(fiber.StatusForbidden, "cannot report a signal for another user")
+	}
+
+	// Default extra to empty JSON object so the inserted column is valid jsonb.
+	extra := b.Extra
+	if len(extra) == 0 {
+		extra = json.RawMessage("{}")
+	}
+
+	source := "client"
+	if b.Type == "one_shot" {
+		source = "admin"
+	}
+
+	var nonce any
+	if b.ClientNonce != "" {
+		nonce = b.ClientNonce
+	}
+
+	if app.writePool == nil {
+		return fiber.NewError(fiber.StatusServiceUnavailable, "write pool not configured")
+	}
+	_, err = app.writePool.Exec(c.Context(), `
+		INSERT INTO challenge_signals (type, user_id, extra, source, client_nonce)
+		VALUES ($1::challenge_signal_type, $2, $3::jsonb, $4, $5)
+		ON CONFLICT (type, user_id, client_nonce)
+		  WHERE client_nonce IS NOT NULL
+		  DO NOTHING
+	`, b.Type, targetUserID, string(extra), source, nonce)
+	if err != nil {
+		return fiber.NewError(fiber.StatusInternalServerError, "failed to record signal: "+err.Error())
+	}
+
+	return c.Status(fiber.StatusAccepted).JSON(fiber.Map{"status": "accepted"})
+}

ddl/migrations/0205_challenge_signals.sql

@@ -0,0 +1,55 @@
+-- challenge_signals: client- and admin-reported events that don't surface
+-- from any on-chain or Solana table on their own. Consumed by the m/o/r/
+-- rv/rd challenge processors in api/jobs/challenges/.
+--
+-- One row per discrete event. Rows are append-only — processors track
+-- their own checkpoint into indexing_checkpoints.
+
+BEGIN;
+
+DO $$ BEGIN
+  CREATE TYPE challenge_signal_type AS ENUM (
+    'mobile_install',
+    'one_shot',
+    'referral'
+  );
+EXCEPTION
+  WHEN duplicate_object THEN NULL;
+END $$;
+
+CREATE TABLE IF NOT EXISTS challenge_signals (
+  id            bigserial PRIMARY KEY,
+  type          challenge_signal_type NOT NULL,
+  user_id       integer NOT NULL,
+  extra         jsonb NOT NULL DEFAULT '{}'::jsonb,
+  created_at    timestamptz NOT NULL DEFAULT now(),
+  source        varchar,
+  client_nonce  varchar
+);
+
+-- Dedupe replays of the same client-reported event.
+CREATE UNIQUE INDEX IF NOT EXISTS challenge_signals_nonce_idx
+  ON challenge_signals (type, user_id, client_nonce)
+  WHERE client_nonce IS NOT NULL;
+
+-- For each processor's incremental scan.
+CREATE INDEX IF NOT EXISTS challenge_signals_type_id_idx
+  ON challenge_signals (type, id);
+
+-- Phase 3 catalog rows.
+INSERT INTO challenges (id, type, amount, active, step_count, starting_block, weekly_pool, cooldown_days) VALUES
+  ('m',  'boolean',   '1', true,  NULL,        25346436, 25000,      7),
+  ('r',  'aggregate', '1', true,  5,           25346436, 25000,      7),
+  ('rv', 'aggregate', '1', true,  5000,        25346436, 25000,      7),
+  ('rd', 'boolean',   '1', true,  NULL,        25346436, 25000,      7),
+  ('o',  'aggregate', '1', true,  2147483647,  0,        2147483647, 0)
+ON CONFLICT (id) DO UPDATE SET
+  type           = EXCLUDED.type,
+  amount         = EXCLUDED.amount,
+  active         = EXCLUDED.active,
+  step_count     = EXCLUDED.step_count,
+  starting_block = EXCLUDED.starting_block,
+  weekly_pool    = EXCLUDED.weekly_pool,
+  cooldown_days  = EXCLUDED.cooldown_days;
+
+COMMIT;

jobs/challenges/processor_test.go

@@ -17,11 +17,20 @@ func withChallengesDB(t *testing.T) *pgxpool.Pool {
 	pool := database.CreateTestDatabase(t, "test_jobs")
 	t.Cleanup(func() { pool.Close() })
 
-	// Seed Phase 1 challenges catalog inline. We don't run the production
-	// migration here because the test_jobs template DB isn't routed through
-	// the ddl runner — keeping the seed local to the test makes intent
-	// clearer too.
 	ctx := context.Background()
+	// The latest pg_migrate preflight seeds a `0x0` block with
+	// is_current=true. database.Seed() also inserts a `block1` block
+	// with is_current=true, which collides on the blocks_is_current_idx
+	// unique partial index. Clear the preflight row so Seed() can
+	// install its expected fixture.
+	if _, err := pool.Exec(ctx, "DELETE FROM blocks WHERE blockhash = '0x0'"); err != nil {
+		t.Fatalf("clean preflight block: %v", err)
+	}
+
+	// Seed Phase 1+2+3 challenges catalog inline. We don't run the
+	// production migration here because the test_jobs template DB isn't
+	// routed through the ddl runner — keeping the seed local to the
+	// test makes intent clearer too.
 	rows := []struct {
 		id, typ, amount       string
 		active                bool
@@ -49,6 +58,12 @@ func withChallengesDB(t *testing.T) *pgxpool.Pool {
 		{"w", "aggregate", "1000", true, i32p(2147483647), 98950182, 50000, i32p(7)},
 		{"b", "aggregate", "1", true, i32p(2147483647), 220157041, 25000, i32p(7)},
 		{"s", "aggregate", "5", true, i32p(2147483647), 220157041, 25000, i32p(7)},
+		// Phase 3 (signal-driven)
+		{"m", "boolean", "1", true, nil, 25346436, 25000, i32p(7)},
+		{"r", "aggregate", "1", true, i32p(5), 25346436, 25000, i32p(7)},
+		{"rv", "aggregate", "1", true, i32p(5000), 25346436, 25000, i32p(7)},
+		{"rd", "boolean", "1", true, nil, 25346436, 25000, i32p(7)},
+		{"o", "aggregate", "1", true, i32p(2147483647), 0, 2147483647, nil},
 	}
 	for _, r := range rows {
 		_, err := pool.Exec(ctx, `