diff --git a/go/internal/ingest/board_webhook.go b/go/internal/ingest/board_webhook.go new file mode 100644 index 000000000..fc14135a8 --- /dev/null +++ b/go/internal/ingest/board_webhook.go @@ -0,0 +1,241 @@ +package ingest + +// The board webhook arm (RIG-2883 T1, design.md:227-302): the second consumer +// behind the one GitHub ingress. The webhook handler fans each accepted event +// to this arm's Enqueue (a non-blocking channel try-send, github_webhook.go:44-51), +// and a single drain goroutine hydrates each changed coordinate via a +// conditional GET and sinks the fresh issue through the shared Ingester — the +// exact poll-path normalization (ingest.go:82-99), never re-implemented. +// +// The webhook payload carries only Number/HTMLURL/State (whIssue, +// githubapp_webhook.go:52-57), not the Title/Body/Labels TranslateIssue maps, +// so the arm HYDRATES on event (OQ-4): one conditional GET per DISTINCT changed +// coordinate. The drain COALESCES per coordinate first (design.md:269-280): an +// edit storm of N rapid events on one issue costs ONE GET, not N — the event +// only proves "changed", so one fresh read serves the whole burst. + +import ( + "context" + "errors" + "expvar" + "log/slog" + "strings" + "sync/atomic" + + "github.com/RigelBuild/compass/go/internal/forge" + compassv1internal "github.com/RigelBuild/compass/go/internal/gen/compass/v1" +) + +// boardWebhookDrops is the exported queue-full/drop metric (design.md:288-292): +// a sustained queue-full silently degrades the hot path to a de-facto 30-min +// poll (the reconciler heal ceiling), so the drop is scrapeable for an alerting +// threshold — a counter+Warn alone is not enough to notice the degradation. +var boardWebhookDrops = expvar.NewInt("compass_board_webhook_drops") + +// defaultBoardQueueSize is the bounded drain-queue depth. Sized to absorb a +// normal edit-storm burst between drains; a sustained overflow (the drain paused +// on ErrBudgetExhausted while events keep arriving) drops with the metric+Warn +// and is healed by the T3 reconciler. +const defaultBoardQueueSize = 1024 + +// issueHydrator is the conditional point-read seam (satisfied structurally by +// *forge.GitHub via GetIssueConditional, notify_reader.go:129). Defined locally +// so this package imports no concrete forge client on the hydrate path. +type issueHydrator interface { + GetIssueConditional(ctx context.Context, repo string, number uint64, etag string) (forge.ConditionalResult[forge.Issue], error) +} + +// TargetChecker gates events to subscribed repos (forge_repo_subscriptions +// WHERE enabled) — the DL-162 target model, point-checked per event. A local +// interface; *store.Store satisfies it at the T5 wiring (this package imports no +// store, ingest.go:7-8). +type TargetChecker interface { + IsEnabledRepo(ctx context.Context, repo string) (bool, error) +} + +// BoardArmConfig configures the board webhook arm. +type BoardArmConfig struct { + // QueueSize is the bounded drain-queue depth; <= 0 uses defaultBoardQueueSize. + QueueSize int + // Log is the arm logger; nil uses slog.Default(). + Log *slog.Logger +} + +// boardCoord is the (repo, number) coalescing key: the drain collapses every +// queued event for one coordinate to a single hydrate GET. +type boardCoord struct { + repo string + number uint64 +} + +// BoardWebhookArm consumes board-relevant forge events from the webhook ingress +// and sinks hydrated issues through the shared Ingester pipeline. +type BoardWebhookArm struct { + queue chan forge.ForgeEvent + hydrator issueHydrator + ing *Ingester + targets TargetChecker + log *slog.Logger + dropped atomic.Int64 +} + +// NewBoardWebhookArm returns an arm that hydrates each accepted event through h, +// gates repos through targets, and sinks through ing. +func NewBoardWebhookArm(h issueHydrator, ing *Ingester, targets TargetChecker, cfg BoardArmConfig) *BoardWebhookArm { + log := cfg.Log + if log == nil { + log = slog.Default() + } + size := cfg.QueueSize + if size <= 0 { + size = defaultBoardQueueSize + } + return &BoardWebhookArm{ + queue: make(chan forge.ForgeEvent, size), + hydrator: h, + ing: ing, + targets: targets, + log: log, + } +} + +// Dropped reports the number of events this arm dropped on a full queue (the +// same fact published to the boardWebhookDrops expvar). Test-observable. +func (a *BoardWebhookArm) Dropped() int64 { return a.dropped.Load() } + +// Enqueue satisfies server.ForgeEventSink's contract (github_webhook.go:44-51): +// it MUST NOT block. It filters to board-relevant issue events (Change ∈ +// {OPENED, STATE, UPDATE} ∧ Kind == ISSUE — PR events and COMMENT-change events +// dropped, design.md:231-237) then channel try-sends; a full queue DROPS the +// event with the drop metric + a Warn (the T3 reconciler heals it). +func (a *BoardWebhookArm) Enqueue(_ context.Context, ev forge.ForgeEvent) { + if !boardRelevant(ev) { + return + } + select { + case a.queue <- ev: + default: + a.dropped.Add(1) + boardWebhookDrops.Add(1) + a.log.Warn("board webhook: queue full, dropping event (reconciler heals)", + "repo", ev.Repo, "number", ev.Number, "change", ev.Change) + } +} + +// boardRelevant reports whether an event is a board-relevant issue change: +// Kind == ISSUE and Change ∈ {OPENED, STATE, UPDATE}. COMMENT-change issue +// events (issue_comment also parses to Kind ISSUE, githubapp_webhook.go:196-210) +// and every PR-kind event are excluded — the board projects issues only, and +// admitting comments would burn one hydrate GET per comment (design.md:231-237). +func boardRelevant(ev forge.ForgeEvent) bool { + if ev.Kind != compassv1internal.ForgeArtifactKind_FORGE_ARTIFACT_KIND_ISSUE { + return false + } + switch ev.Change { + case compassv1internal.ForgeNotificationKind_FORGE_NOTIFICATION_KIND_OPENED, + compassv1internal.ForgeNotificationKind_FORGE_NOTIFICATION_KIND_STATE, + compassv1internal.ForgeNotificationKind_FORGE_NOTIFICATION_KIND_UPDATE: + return true + default: + return false + } +} + +// Run drains the queue until ctx is cancelled — then it returns nil (clean +// shutdown, driver.go:95-99 idiom). Each drain COALESCES per coordinate before +// hydrating, so an N-event burst on one issue costs one GET. +func (a *BoardWebhookArm) Run(ctx context.Context) error { + for { + select { + case <-ctx.Done(): + return nil + case ev := <-a.queue: + a.drainBatch(ctx, ev) + } + } +} + +// drainBatch coalesces first into a distinct-coordinate set (design.md:271-274): +// it seeds with the event that woke the drain, non-blockingly drains every other +// currently-queued event, and keys each on its NORMALIZED (repo, number) — so an +// edit storm on one issue, and a mixed-case duplicate of one repo, both collapse +// to a single coordinate. It then hydrates + sinks each distinct coordinate once, +// in arrival order. +func (a *BoardWebhookArm) drainBatch(ctx context.Context, first forge.ForgeEvent) { + seen := map[boardCoord]struct{}{} + var order []boardCoord + + add := func(ev forge.ForgeEvent) { + c := boardCoord{repo: normalizeBoardRepo(ev.Repo), number: ev.Number} + if _, ok := seen[c]; ok { + return + } + seen[c] = struct{}{} + order = append(order, c) + } + + add(first) + for { + select { + case ev := <-a.queue: + add(ev) + default: + goto process + } + } + +process: + for _, c := range order { + if err := a.hydrateAndSink(ctx, c); err != nil { + if errors.Is(err, forge.ErrBudgetExhausted) { + // Budget exhausted pauses the drain: abandon the rest of this + // batch (the reconciler heals the un-hydrated coordinates), and + // the next batch resumes once the client gate reopens + // (github.go:83-102) — the reconciler's treatment + // (notify_reconcile.go:139-142). + a.log.WarnContext(ctx, "board webhook: budget exhausted, pausing drain (reconciler heals)", + "repo", c.repo, "number", c.number) + return + } + // Per-event errors log-and-continue (driver.go:96-98 idiom). + a.log.WarnContext(ctx, "board webhook: hydrate/sink failed (isolated)", + "repo", c.repo, "number", c.number, "err", err) + } + if ctx.Err() != nil { + return + } + } +} + +// hydrateAndSink gates the coordinate's repo, hydrates the issue via an +// unconditional conditional GET (the event proves change; no stored per-issue +// ETag in v1, design.md:276-277), and sinks the fresh issue through the shared +// Ingester (ingest.go:82-99 — the one owner-strip/translate/stamp pipeline). A +// non-enabled repo is dropped silently. +func (a *BoardWebhookArm) hydrateAndSink(ctx context.Context, c boardCoord) error { + enabled, err := a.targets.IsEnabledRepo(ctx, c.repo) + if err != nil { + return err + } + if !enabled { + return nil + } + res, err := a.hydrator.GetIssueConditional(ctx, c.repo, c.number, "") + if err != nil { + return err + } + if res.NotModified { + // Unreachable with an empty ETag (a 200-equivalent); defensively skip. + return nil + } + return a.ing.IngestIssues(ctx, c.repo, []forge.Issue{res.V}) +} + +// normalizeBoardRepo lowercases the event repo at the boundary (Global +// Constraint 8, design.md:213-225): ParseGitHubEvent sets Repo from the +// case-PRESERVED payload full_name, while subscription rows and board +// coordinates are lowercased at the seed/upsert boundary — so a raw event repo +// would silently miss the IsEnabledRepo row and mint a duplicate coordinate. +func normalizeBoardRepo(raw string) string { + return strings.ToLower(strings.TrimSpace(raw)) +} diff --git a/go/internal/ingest/board_webhook_test.go b/go/internal/ingest/board_webhook_test.go new file mode 100644 index 000000000..5268a8c14 --- /dev/null +++ b/go/internal/ingest/board_webhook_test.go @@ -0,0 +1,303 @@ +package ingest + +import ( + "context" + "strings" + "sync" + "testing" + + "github.com/RigelBuild/compass/go/internal/forge" + compassv1internal "github.com/RigelBuild/compass/go/internal/gen/compass/v1" +) + +// fakeHydrator is a scripted issueHydrator that records every GetIssueConditional +// call (repo+number) so a test can assert the coalesce lever at the call count. +type fakeHydrator struct { + mu sync.Mutex + calls []boardCoord + result forge.Issue + err error + errOnce bool // if true, return err only on the first call, then succeed + served int +} + +func (h *fakeHydrator) GetIssueConditional(_ context.Context, repo string, number uint64, _ string) (forge.ConditionalResult[forge.Issue], error) { + h.mu.Lock() + defer h.mu.Unlock() + h.calls = append(h.calls, boardCoord{repo: repo, number: number}) + h.served++ + if h.err != nil && (!h.errOnce || h.served == 1) { + return forge.ConditionalResult[forge.Issue]{}, h.err + } + res := h.result + res.Number = number + return forge.ConditionalResult[forge.Issue]{V: res}, nil +} + +func (h *fakeHydrator) callCount() int { + h.mu.Lock() + defer h.mu.Unlock() + return len(h.calls) +} + +// fakeTargets is a scripted TargetChecker: a repo is enabled iff it is in the +// lowercased enabled set. It records every checked repo. +type fakeTargets struct { + mu sync.Mutex + enabled map[string]bool + checked []string + checkErr error +} + +func newFakeTargets(repos ...string) *fakeTargets { + m := map[string]bool{} + for _, r := range repos { + m[r] = true + } + return &fakeTargets{enabled: m} +} + +func (f *fakeTargets) IsEnabledRepo(_ context.Context, repo string) (bool, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.checked = append(f.checked, repo) + if f.checkErr != nil { + return false, f.checkErr + } + return f.enabled[repo], nil +} + +// newArmHarness wires a fake hydrator + targets + a recording sink into a +// BoardWebhookArm with a real Ingester (so the full owner-strip/translate/stamp +// pipeline runs) and a small queue for the full-queue test. +func newArmHarness(t *testing.T, h issueHydrator, targets TargetChecker, queueSize int) (*BoardWebhookArm, *recordingSink) { + t.Helper() + sink := &recordingSink{} + // The Ingester's forgeReader is unused on the webhook path (IngestIssues + // takes caller-fetched issues), so a bare fake provider suffices. + ing := NewIngester(forge.NewFakeProvider("gh"), sink, testForgeRef()) + arm := NewBoardWebhookArm(h, ing, targets, BoardArmConfig{QueueSize: queueSize}) + return arm, sink +} + +func issueEvent(repo string, number uint64, change compassv1internal.ForgeNotificationKind) forge.ForgeEvent { + return forge.ForgeEvent{ + Repo: repo, + Number: number, + Kind: compassv1internal.ForgeArtifactKind_FORGE_ARTIFACT_KIND_ISSUE, + Change: change, + } +} + +const ( + changeUpdate = compassv1internal.ForgeNotificationKind_FORGE_NOTIFICATION_KIND_UPDATE + changeState = compassv1internal.ForgeNotificationKind_FORGE_NOTIFICATION_KIND_STATE + changeComment = compassv1internal.ForgeNotificationKind_FORGE_NOTIFICATION_KIND_COMMENT + boardKindPR = compassv1internal.ForgeArtifactKind_FORGE_ARTIFACT_KIND_PULL_REQUEST +) + +// drainAll deterministically processes every currently-queued event by pulling +// one head event and letting drainBatch coalesce+drain the rest — no goroutine, +// no clock. It mirrors exactly one Run iteration per queued head, so the result +// is the same as Run would produce, but synchronously observable. +func drainAll(ctx context.Context, arm *BoardWebhookArm) { + for { + select { + case ev := <-arm.queue: + arm.drainBatch(ctx, ev) + default: + return + } + } +} + +// TestArmHydratesAndSinks: a board-relevant issue event hydrates the coordinate +// and the fresh issue reaches PublishIssueUpdate with a stripped body and a +// stamped ForgeRef (the full pipeline ran). +func TestArmHydratesAndSinks(t *testing.T) { + body := stamped(t, "real body", forge.Author{AgentHandle: "atlas", OwnerHandle: "matt", SessionID: "s1"}) + h := &fakeHydrator{result: forge.Issue{Title: "a bug", Body: body, State: "open"}} + arm, sink := newArmHarness(t, h, newFakeTargets("owner/repo"), 16) + + arm.Enqueue(context.Background(), issueEvent("owner/repo", 7, changeUpdate)) + drainAll(context.Background(), arm) + + if len(sink.got) != 1 { + t.Fatalf("sink got %d issues, want 1", len(sink.got)) + } + got := sink.got[0] + if got.GetNumber() != 7 { + t.Errorf("Number = %d, want 7", got.GetNumber()) + } + if got.GetBody() != "real body" { + t.Errorf("Body = %q, want stripped %q", got.GetBody(), "real body") + } + if got.GetAgent().GetAgentHandle() != "atlas" { + t.Errorf("Agent.AgentHandle = %q, want atlas", got.GetAgent().GetAgentHandle()) + } + if got.GetForge().GetProvider() != testForgeRef().GetProvider() || got.GetRepo() != "owner/repo" { + t.Errorf("ForgeRef/Repo not stamped: forge=%v repo=%q", got.GetForge(), got.GetRepo()) + } + if h.callCount() != 1 { + t.Errorf("hydrate calls = %d, want 1", h.callCount()) + } +} + +// TestArmDropsNonEnabledRepo: an event for a repo with no enabled subscription +// row is gated out before the hydrate — no sink, no hydrate GET spent. +func TestArmDropsNonEnabledRepo(t *testing.T) { + h := &fakeHydrator{result: forge.Issue{Title: "x", State: "open"}} + arm, sink := newArmHarness(t, h, newFakeTargets(), 16) + + arm.Enqueue(context.Background(), issueEvent("owner/repo", 7, changeUpdate)) + drainAll(context.Background(), arm) + + if len(sink.got) != 0 { + t.Fatalf("sink got %d, want 0 for a non-enabled repo", len(sink.got)) + } + if h.callCount() != 0 { + t.Fatalf("hydrate calls = %d, want 0 (gate before hydrate)", h.callCount()) + } +} + +// TestArmDropsPRKind: a pull_request-kind event is filtered at Enqueue — never +// queued, never hydrated. +func TestArmDropsPRKind(t *testing.T) { + h := &fakeHydrator{result: forge.Issue{State: "open"}} + arm, sink := newArmHarness(t, h, newFakeTargets("owner/repo"), 16) + + ev := issueEvent("owner/repo", 7, changeUpdate) + ev.Kind = boardKindPR + arm.Enqueue(context.Background(), ev) + + if l := len(arm.queue); l != 0 { + t.Fatalf("queue len = %d, want 0 (PR-kind filtered at Enqueue)", l) + } + drainAll(context.Background(), arm) + if len(sink.got) != 0 || h.callCount() != 0 { + t.Fatalf("PR-kind leaked: sink=%d hydrate=%d", len(sink.got), h.callCount()) + } +} + +// TestArmDropsCommentChange: a COMMENT-change issue event is filtered at Enqueue +// — no hydrate GET is spent (the cost bound, design.md:233-237). +func TestArmDropsCommentChange(t *testing.T) { + h := &fakeHydrator{result: forge.Issue{State: "open"}} + arm, sink := newArmHarness(t, h, newFakeTargets("owner/repo"), 16) + + arm.Enqueue(context.Background(), issueEvent("owner/repo", 7, changeComment)) + if l := len(arm.queue); l != 0 { + t.Fatalf("queue len = %d, want 0 (COMMENT-change filtered)", l) + } + drainAll(context.Background(), arm) + if len(sink.got) != 0 || h.callCount() != 0 { + t.Fatalf("COMMENT-change leaked: sink=%d hydrate=%d", len(sink.got), h.callCount()) + } +} + +// TestArmNormalizesMixedCaseRepo: a mixed-case full_name is lowercased before +// the gate and sink, so the lowercased subscription row matches; a same-issue +// mixed-case duplicate coalesces to one coordinate (one hydrate GET, one sink). +func TestArmNormalizesMixedCaseRepo(t *testing.T) { + h := &fakeHydrator{result: forge.Issue{State: "open"}} + targets := newFakeTargets("owner/repo") + arm, sink := newArmHarness(t, h, targets, 16) + + // Two events on the same coordinate, differing only by repo case. + arm.Enqueue(context.Background(), issueEvent("Owner/Repo", 7, changeUpdate)) + arm.Enqueue(context.Background(), issueEvent("owner/repo", 7, changeState)) + drainAll(context.Background(), arm) + + if len(sink.got) != 1 { + t.Fatalf("sink got %d, want 1 (mixed-case dup coalesced)", len(sink.got)) + } + if h.callCount() != 1 { + t.Fatalf("hydrate calls = %d, want 1 (one coordinate)", h.callCount()) + } + if got := sink.got[0].GetRepo(); got != "owner/repo" { + t.Errorf("sinked Repo = %q, want lowercased owner/repo", got) + } + for _, r := range targets.checked { + if r != strings.ToLower(r) { + t.Errorf("IsEnabledRepo checked non-normalized repo %q", r) + } + } +} + +// TestArmFullQueueDropsWithoutBlocking: with the drain not running, a queue of +// capacity 1 accepts one event and DROPS the second without blocking Enqueue. +func TestArmFullQueueDropsWithoutBlocking(t *testing.T) { + h := &fakeHydrator{result: forge.Issue{State: "open"}} + arm, _ := newArmHarness(t, h, newFakeTargets("owner/repo"), 1) + + arm.Enqueue(context.Background(), issueEvent("owner/repo", 1, changeUpdate)) + // Second enqueue must not block (drain is not running) and must drop. + done := make(chan struct{}) + go func() { + arm.Enqueue(context.Background(), issueEvent("owner/repo", 2, changeUpdate)) + close(done) + }() + <-done // gates on Enqueue returning; it hangs the test (no close) if it blocks + if arm.Dropped() != 1 { + t.Fatalf("Dropped() = %d, want 1", arm.Dropped()) + } +} + +// TestArmHydrateErrorRetriesNextEvent: a hydrate error on one coordinate is +// isolated (log-and-continue) and never crashes the drain — a subsequent event +// on a different coordinate is hydrated and sinked. +func TestArmHydrateErrorRetriesNextEvent(t *testing.T) { + h := &fakeHydrator{result: forge.Issue{State: "open"}, err: errBoom, errOnce: true} + arm, sink := newArmHarness(t, h, newFakeTargets("owner/repo"), 16) + + arm.Enqueue(context.Background(), issueEvent("owner/repo", 1, changeUpdate)) + drainAll(context.Background(), arm) + // First coordinate errored: nothing sinked, drain did not crash. + if len(sink.got) != 0 { + t.Fatalf("sink got %d after the erroring event, want 0", len(sink.got)) + } + // A second event succeeds. + arm.Enqueue(context.Background(), issueEvent("owner/repo", 2, changeUpdate)) + drainAll(context.Background(), arm) + if len(sink.got) != 1 { + t.Fatalf("sink got %d, want 1 (recovered after error)", len(sink.got)) + } + if sink.got[0].GetNumber() != 2 { + t.Errorf("sinked Number = %d, want 2", sink.got[0].GetNumber()) + } +} + +// TestArmCoalescesBurst: a burst of N events on ONE coordinate, all queued +// before the drain wakes, coalesces to ONE hydrate GET (the coalesce lever, +// observable at the fake hydrator call count). +func TestArmCoalescesBurst(t *testing.T) { + h := &fakeHydrator{result: forge.Issue{State: "open"}} + arm, sink := newArmHarness(t, h, newFakeTargets("owner/repo"), 64) + + for range 20 { + arm.Enqueue(context.Background(), issueEvent("owner/repo", 7, changeUpdate)) + } + drainAll(context.Background(), arm) + + if h.callCount() != 1 { + t.Fatalf("hydrate calls = %d, want 1 (20-event burst coalesced to one GET)", h.callCount()) + } + if len(sink.got) != 1 { + t.Fatalf("sink got %d, want 1", len(sink.got)) + } +} + +// TestArmRunReturnsNilOnCancel: Run drains until ctx cancel and returns nil +// (driver.go:95-99 idiom). Gated on the result channel, not a clock. +func TestArmRunReturnsNilOnCancel(t *testing.T) { + h := &fakeHydrator{result: forge.Issue{State: "open"}} + arm, _ := newArmHarness(t, h, newFakeTargets("owner/repo"), 16) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- arm.Run(ctx) }() + cancel() + if err := <-done; err != nil { + t.Fatalf("Run returned %v, want nil on cancel", err) + } +}