Score Poller Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Add a dedicated goroutine that polls live scores from the Events bookie rows on a short interval, independent of market update notifications, so goal/score changes are pushed to WebSocket clients within seconds even when bookmakers suspend markets. Architecture: A new pollScores goroutine runs on a fixed ticker (default 5 s, env-tunable). It queries all matches rows with bookie = 'Events' and status = 'live' (or recently-settled), diffs against in-memory state already maintained by lastScoreSig/lastStatus, and broadcasts only changed rows. It reuses all existing diff logic in fetchEventStateUpdates — the only new code is the query that fetches all live Events rows instead of a specific batch of match IDs. No schema changes needed; no new tables. Tech Stack: Go, database/sql, existing database.Postgres pool, existing models.UpdateItem / dto.Score types, existing broadcastUpdates + Hub fan-out pipeline.

Global Constraints

No new dependencies — reuse database.Postgres, existing types, utils.SafeGoWithRestart.
Follow existing env-var kill-switch pattern (os.Getenv read deferred until runtime, not init).
All new public-surface behaviour must have a unit test; DB-touching functions are tested through existing mock patterns (pure-logic extraction where needed).
No comments that restate what the code does — only non-obvious “why” comments.
Run go test ./websocket/... before every commit; must pass green.

File Map

File	Change
`v3/websocket/poller_scores.go`	Add `fetchAllLiveEventStateUpdates(ctx)` — queries all live Events rows
`v3/websocket/poller.go`	Add `scorePollInterval` field + `pollScores()` goroutine; wire into `Start()`/`Stop()`
`v3/websocket/poller_scores_test.go`	Add tests for `fetchAllLiveEventStateUpdates` query shape and `pollScoreInterval` env parsing

Task 1: Add `fetchAllLiveEventStateUpdates` to poller_scores.go

This is the query that fetches every live Events row instead of a specific batch. It reuses the existing per-row diff logic inside fetchEventStateUpdates — we extract the inner loop into a shared helper so both callers stay DRY. Files:

Modify: v3/websocket/poller_scores.go
Test: v3/websocket/poller_scores_test.go

Interfaces:

Produces: func (p *Poller) fetchAllLiveEventStateUpdates(ctx context.Context) []*models.UpdateItem

The method signature matches fetchEventStateUpdates except it takes no matchIDs slice — it queries all live rows itself.

Why this query shape

SELECT id, sport, league, date, status, scores_home, scores_away, period_scores, created_at
FROM matches
WHERE bookie = 'Events'
  AND status IN ('live', 'settled', 'cancelled')
  AND date >= NOW() - INTERVAL '4 hours'

bookie = 'Events' — scores are stored only on the synthetic Events row, never on bookmaker rows.
status IN ('live', 'settled', 'cancelled') — we still want to push a settled/cancelled transition; prematch rows are excluded so the query stays small.
date >= NOW() - INTERVAL '4 hours' — belt-and-suspenders cap so a stale prematch row that somehow has status=‘live’ doesn’t pollute results; 4 h covers any match that kicked off recently and might still be live.

This query typically returns O(tens) of rows during peak load (only concurrent live matches), so it is cheap.

Step 1: Write the failing test

Add to v3/websocket/poller_scores_test.go:

func TestFetchAllLiveEventStateUpdates_ReturnsNilWhenNoSubscribers(t *testing.T) {
    // Hub with zero subscribers should skip DB and return nil.
    hub := &Hub{clients: make(map[string][]*models.WebSocketClient)}
    p := &Poller{
        hub:          hub,
        lastScoreSig: make(map[string]string),
        lastStatus:   make(map[string]string),
    }
    ctx := context.Background()
    got := p.fetchAllLiveEventStateUpdates(ctx)
    if got != nil {
        t.Fatalf("expected nil, got %v", got)
    }
}

Step 2: Run test to verify it fails

cd v3 && go test ./websocket/... -run TestFetchAllLiveEventStateUpdates -v

Expected: compile error — fetchAllLiveEventStateUpdates undefined.

Step 3: Implement fetchAllLiveEventStateUpdates in poller_scores.go

Add at the bottom of v3/websocket/poller_scores.go, after the existing nullStr function:

// fetchAllLiveEventStateUpdates queries every live/recently-settled Events row
// and emits score/status UpdateItems for any that changed since last broadcast.
// Called by the dedicated score-poller goroutine independently of market notifications.
func (p *Poller) fetchAllLiveEventStateUpdates(ctx context.Context) []*models.UpdateItem {
    wantScores := p.hub.HasChannelSubscribers(models.ChannelScores)
    wantStatus := p.hub.HasChannelSubscribers(models.ChannelStatus)
    if !wantScores && !wantStatus {
        return nil
    }

    rows, err := database.Postgres.Query(ctx,
        `SELECT id, sport, league, date, status, scores_home, scores_away, period_scores, created_at
         FROM matches
         WHERE bookie = $1
           AND status IN ('live', 'settled', 'cancelled')
           AND date >= NOW() - INTERVAL '4 hours'`,
        eventsBookie)
    if err != nil {
        return nil
    }
    defer rows.Close()

    now := time.Now()
    var updates []*models.UpdateItem
    for rows.Next() {
        var id string
        var sport, league, status, scoresHome, scoresAway, periodScores sql.NullString
        var date time.Time
        var createdAt sql.NullTime
        if err := rows.Scan(&id, &sport, &league, &date, &status, &scoresHome, &scoresAway, &periodScores, &createdAt); err != nil {
            continue
        }

        statusStr := status.String
        hasScore := scoresHome.Valid && scoresHome.String != "" && scoresAway.Valid && scoresAway.String != ""
        scoreSig := scoresHome.String + "|" + scoresAway.String + "|" + periodScores.String

        p.scoreStateMu.Lock()
        statusChanged := statusStr != "" && p.lastStatus[id] != statusStr
        scoreChanged := p.lastScoreSig[id] != scoreSig
        if statusChanged {
            p.lastStatus[id] = statusStr
        }
        if scoreChanged {
            p.lastScoreSig[id] = scoreSig
        }
        if len(p.lastScoreSig) > scoreStateCap {
            evictHalf(p.lastScoreSig)
            evictHalf(p.lastStatus)
        }
        p.scoreStateMu.Unlock()

        emitScore := scoreChanged && wantScores && hasScore
        emitStatus := shouldEmitStatus(statusStr, statusChanged, wantStatus, createdAt, now)
        if !emitScore && !emitStatus {
            continue
        }

        base := models.UpdateItem{
            MatchID:     id,
            BaseEventID: id,
            UpdatedAt:   now,
            Sport:       nullStr(sport),
            League:      nullStr(league),
            Date:        &date,
            Status:      nullStr(status),
            Scores:      buildScore(scoresHome, scoresAway, periodScores),
        }
        if emitScore {
            u := base
            u.Type = models.UpdateTypeScore
            updates = append(updates, &u)
        }
        if emitStatus {
            u := base
            u.Type = models.UpdateTypeStatus
            updates = append(updates, &u)
        }
    }
    return updates
}

Step 4: Run test to verify it passes

cd v3 && go test ./websocket/... -run TestFetchAllLiveEventStateUpdates -v

Expected: PASS

Step 5: Commit

git add v3/websocket/poller_scores.go v3/websocket/poller_scores_test.go
git commit -m "feat(ws): fetchAllLiveEventStateUpdates — query all live Events rows for score poller"

Task 2: Add `pollScores` goroutine to poller.go

Wire the new query into a dedicated goroutine with a configurable interval and clean shutdown. The goroutine must:

Check for subscribers before hitting DB (cheap guard)
Log only when it actually broadcasts something (avoids logspam)
Be skippable via env var for emergencies

Files:

Modify: v3/websocket/poller.go
Test: v3/websocket/poller_scores_test.go

Interfaces:

Consumes: p.fetchAllLiveEventStateUpdates(ctx) []*models.UpdateItem (Task 1)
Consumes: p.broadcastUpdates(updates) (existing)

Env var: `WS_SCORE_POLL_INTERVAL`

Read as a time.Duration string (e.g. "5s", "10s"). Default: 5s. Set to "0" or "off" to disable the goroutine entirely (kill switch for debugging). This follows the same deferred-read pattern as WS_COMPRESS in hub.go — read inside the goroutine, not at struct init, so env is loaded before it’s consumed.

Step 1: Write the failing test for interval parsing

Add to v3/websocket/poller_scores_test.go:

func TestScorePollInterval(t *testing.T) {
    cases := []struct {
        env  string
        want time.Duration
        skip bool
    }{
        {"", 5 * time.Second, false},
        {"10s", 10 * time.Second, false},
        {"0", 0, true},
        {"off", 0, true},
        {"garbage", 5 * time.Second, false}, // falls back to default
    }
    for _, c := range cases {
        t.Run(c.env, func(t *testing.T) {
            t.Setenv("WS_SCORE_POLL_INTERVAL", c.env)
            got, skip := scorePollInterval()
            if skip != c.skip {
                t.Fatalf("skip: got %v want %v", skip, c.skip)
            }
            if !skip && got != c.want {
                t.Fatalf("interval: got %v want %v", got, c.want)
            }
        })
    }
}

Step 2: Run test to verify it fails

cd v3 && go test ./websocket/... -run TestScorePollInterval -v

Expected: compile error — scorePollInterval undefined.

Step 3: Add scorePollInterval() helper and pollScores() to poller.go

Add near the bottom of v3/websocket/poller.go, before the closing of the file:

// scorePollInterval reads WS_SCORE_POLL_INTERVAL and returns the polling
// interval plus a skip flag. "0" or "off" disables the goroutine entirely.
func scorePollInterval() (time.Duration, bool) {
    const defaultInterval = 5 * time.Second
    v := os.Getenv("WS_SCORE_POLL_INTERVAL")
    if v == "0" || v == "off" {
        return 0, true
    }
    if v == "" {
        return defaultInterval, false
    }
    d, err := time.ParseDuration(v)
    if err != nil || d <= 0 {
        log.Printf("[SCORE-POLL] Invalid WS_SCORE_POLL_INTERVAL %q, using default %v", v, defaultInterval)
        return defaultInterval, false
    }
    return d, true
}

// pollScores runs on a fixed ticker and pushes score/status changes to clients
// without relying on market-update notifications. This closes the gap where
// bookmakers suspend markets after a goal, which previously caused 10+ minute
// score-push delays.
func (p *Poller) pollScores() {
    interval, skip := scorePollInterval()
    if skip {
        log.Println("[SCORE-POLL] Disabled via WS_SCORE_POLL_INTERVAL=0/off")
        return
    }
    log.Printf("[SCORE-POLL] Starting with interval %v", interval)

    ticker := time.NewTicker(interval)
    defer ticker.Stop()

    for {
        select {
        case <-p.ctx.Done():
            return
        case <-ticker.C:
            ctx, cancel := context.WithTimeout(p.ctx, interval)
            updates := p.fetchAllLiveEventStateUpdates(ctx)
            cancel()
            if len(updates) > 0 {
                log.Printf("[SCORE-POLL] Broadcasting %d score/status update(s)", len(updates))
                p.broadcastUpdates(updates)
            }
        }
    }
}

Step 4: Wire pollScores into Start()

In v3/websocket/poller.go, in the Start() function, add after the existing SafeGoWithRestart calls:

utils.SafeGoWithRestart("poller-pollScores", p.pollScores, 5*time.Second, 0)

The full Start() body should look like:

func (p *Poller) Start() {
    log.Println("Starting LISTEN/NOTIFY WebSocket poller...")

    if err := p.listener.Listen("market_updates"); err != nil {
        log.Fatalf("Failed to listen on market_updates: %v", err)
    }

    if err := p.listener.Listen("match_updates"); err != nil {
        log.Fatalf("Failed to listen on match_updates: %v", err)
    }

    log.Println("LISTEN/NOTIFY active on market_updates and match_updates")

    utils.SafeGoWithRestart("poller-handleNotifications", p.handleNotifications, 5*time.Second, 0)
    utils.SafeGoWithRestart("poller-processBatches", p.processBatches, 5*time.Second, 0)
    utils.SafeGoWithRestart("poller-backupPolling", p.backupPolling, 5*time.Second, 0)
    utils.SafeGoWithRestart("poller-warmCaches", p.warmCaches, 5*time.Second, 0)
    utils.SafeGoWithRestart("poller-monitorMetrics", p.monitorMetrics, 5*time.Second, 0)
    utils.SafeGoWithRestart("poller-pollScores", p.pollScores, 5*time.Second, 0)
}

Step 5: Run all tests

cd v3 && go test ./websocket/... -v 2>&1 | tail -20

Expected: all PASS, including TestScorePollInterval.

Step 6: Build check

cd v3 && go build ./...

Expected: no errors.

Step 7: Commit

git add v3/websocket/poller.go v3/websocket/poller_scores_test.go
git commit -m "feat(ws): dedicated score-poller goroutine — push goals within 5 s without market notifications"

Self-Review

Spec coverage:

Score/status delays when markets are suspended → covered by pollScores polling all live Events rows every 5 s
High performance → query limited to live + recent rows (O(tens) rows, cheap); subscriber guard skips DB entirely when no one is subscribed
Kill switch → WS_SCORE_POLL_INTERVAL=0 disables goroutine
No duplicate broadcasts → reuses existing lastScoreSig/lastStatus diff maps, same locks
No memory leaks → goroutine exits on ctx.Done(), ticker deferred Stop

Placeholder scan: None found. Type consistency: fetchAllLiveEventStateUpdates returns []*models.UpdateItem — same as fetchEventStateUpdates, consumed by broadcastUpdates which already accepts that type. ✓

Getting Started

Guides

Examples

2026 06 25 score poller

Score Poller Implementation Plan

Global Constraints

File Map

Task 1: Add `fetchAllLiveEventStateUpdates` to poller_scores.go

Why this query shape

Task 2: Add `pollScores` goroutine to poller.go

Env var: `WS_SCORE_POLL_INTERVAL`

Self-Review

​Score Poller Implementation Plan

​Global Constraints

​File Map

​Task 1: Add fetchAllLiveEventStateUpdates to poller_scores.go

​Why this query shape

​Task 2: Add pollScores goroutine to poller.go

​Env var: WS_SCORE_POLL_INTERVAL

​Self-Review

Score Poller Implementation Plan

Global Constraints

File Map

Task 1: Add `fetchAllLiveEventStateUpdates` to poller_scores.go

Why this query shape

Task 2: Add `pollScores` goroutine to poller.go

Env var: `WS_SCORE_POLL_INTERVAL`

Self-Review