Skip to content

Commit 87be1a3

Browse files
tstrombergtstromberg
authored
Merge pull request #44 from codeGROOVE-dev/reliable
Add closed PRs to the polling
2 parents ec25791 + 9c255f6 commit 87be1a3

File tree

4 files changed

+275
-5
lines changed

4 files changed

+275
-5
lines changed

‎go.mod‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@ require (
77
github.com/codeGROOVE-dev/gh-mailto v0.0.0-20251019162917-c3412c017b1f
88
github.com/codeGROOVE-dev/gsm v0.0.0-20251019065141-833fe2363d22
99
github.com/codeGROOVE-dev/retry v1.2.0
10-
github.com/codeGROOVE-dev/sprinkler v0.0.0-20251019110134-896b678fd945
10+
github.com/codeGROOVE-dev/sprinkler v0.0.0-20251020064313-f606185b6b98
1111
github.com/codeGROOVE-dev/turnclient v0.0.0-20251018202306-7cdc0d51856e
1212
github.com/golang-jwt/jwt/v5 v5.3.0
1313
github.com/google/go-github/v50 v50.2.0

‎go.sum‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,8 @@ github.com/codeGROOVE-dev/prx v0.0.0-20251016165946-00c6c6e90c29 h1:MSBy3Ywr3ky/
2222
github.com/codeGROOVE-dev/prx v0.0.0-20251016165946-00c6c6e90c29/go.mod h1:7qLbi18baOyS8yO/6/64SBIqtyzSzLFdsDST15NPH3w=
2323
github.com/codeGROOVE-dev/retry v1.2.0 h1:xYpYPX2PQZmdHwuiQAGGzsBm392xIMl4nfMEFApQnu8=
2424
github.com/codeGROOVE-dev/retry v1.2.0/go.mod h1:8OgefgV1XP7lzX2PdKlCXILsYKuz6b4ZpHa/20iLi8E=
25-
github.com/codeGROOVE-dev/sprinkler v0.0.0-20251019110134-896b678fd945 h1:uZhbGjrIEYfs6Bq2PQgbbtag5gAjMp/NQZGAQsL73m4=
26-
github.com/codeGROOVE-dev/sprinkler v0.0.0-20251019110134-896b678fd945/go.mod h1:/kd3ncsRNldD0MUpbtp5ojIzfCkyeXB7JdOrpuqG7Gg=
25+
github.com/codeGROOVE-dev/sprinkler v0.0.0-20251020064313-f606185b6b98 h1:unjiIF1rx/QZfcTEW/n6EJjde1yd3b1ZbjrWee2Afj4=
26+
github.com/codeGROOVE-dev/sprinkler v0.0.0-20251020064313-f606185b6b98/go.mod h1:/kd3ncsRNldD0MUpbtp5ojIzfCkyeXB7JdOrpuqG7Gg=
2727
github.com/codeGROOVE-dev/turnclient v0.0.0-20251018202306-7cdc0d51856e h1:3qoY6h8SgoeNsIYRM7P6PegTXAHPo8OSOapUunVP/Gs=
2828
github.com/codeGROOVE-dev/turnclient v0.0.0-20251018202306-7cdc0d51856e/go.mod h1:fYwtN9Ql6lY8t2WvCfENx+mP5FUwjlqwXCLx9CVLY20=
2929
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=

‎internal/bot/polling.go‎

Lines changed: 161 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import (
55
"errors"
66
"fmt"
77
"log/slog"
8+
"strings"
89
"time"
910

1011
"github.com/codeGROOVE-dev/slacker/internal/github"
@@ -49,7 +50,7 @@ func (c *Coordinator) PollAndReconcile(ctx context.Context) {
4950
"pr_count", len(prs),
5051
"will_check_each", true)
5152

52-
// Process each PR
53+
// Process each open PR
5354
successCount := 0
5455
errorCount := 0
5556

@@ -93,9 +94,73 @@ func (c *Coordinator) PollAndReconcile(ctx context.Context) {
9394
}
9495
}
9596

97+
// Query closed/merged PRs in last hour to update existing threads
98+
closedPRs, err := gqlClient.ListClosedPRs(ctx, org, 1)
99+
if err != nil {
100+
slog.Warn("failed to poll closed PRs",
101+
"org", org,
102+
"error", err,
103+
"impact", "will retry next poll")
104+
} else {
105+
slog.Info("poll retrieved closed/merged PRs",
106+
"org", org,
107+
"pr_count", len(closedPRs),
108+
"will_update_threads", true)
109+
110+
closedSuccessCount := 0
111+
closedErrorCount := 0
112+
113+
for i := range closedPRs {
114+
pr := &closedPRs[i]
115+
116+
// Create event key for this PR state change
117+
eventKey := fmt.Sprintf("poll_closed:%s:%s:%s", pr.URL, pr.State, pr.UpdatedAt.Format(time.RFC3339))
118+
119+
// Skip if already processed
120+
if c.stateStore.WasProcessed(eventKey) {
121+
slog.Debug("skipping closed PR - already processed",
122+
"pr", fmt.Sprintf("%s/%s#%d", pr.Owner, pr.Repo, pr.Number),
123+
"state", pr.State)
124+
closedSuccessCount++
125+
continue
126+
}
127+
128+
// Update thread for this closed/merged PR
129+
if err := c.updateClosedPRThread(ctx, pr); err != nil {
130+
slog.Warn("failed to update closed PR thread",
131+
"pr", fmt.Sprintf("%s/%s#%d", pr.Owner, pr.Repo, pr.Number),
132+
"state", pr.State,
133+
"error", err)
134+
closedErrorCount++
135+
} else {
136+
// Mark as processed
137+
if err := c.stateStore.MarkProcessed(eventKey, 24*time.Hour); err != nil {
138+
slog.Warn("failed to mark closed PR event as processed",
139+
"pr", fmt.Sprintf("%s/%s#%d", pr.Owner, pr.Repo, pr.Number),
140+
"error", err)
141+
}
142+
closedSuccessCount++
143+
}
144+
145+
// Rate limit
146+
select {
147+
case <-ctx.Done():
148+
slog.Info("polling canceled during closed PR processing", "org", org)
149+
return
150+
case <-time.After(100 * time.Millisecond):
151+
}
152+
}
153+
154+
slog.Info("closed PR processing complete",
155+
"org", org,
156+
"total_closed_prs", len(closedPRs),
157+
"updated", closedSuccessCount,
158+
"errors", closedErrorCount)
159+
}
160+
96161
slog.Info("poll cycle complete",
97162
"org", org,
98-
"total_prs", len(prs),
163+
"total_open_prs", len(prs),
99164
"processed", successCount,
100165
"errors", errorCount,
101166
"next_poll", "5m")
@@ -180,6 +245,100 @@ func (c *Coordinator) reconcilePR(ctx context.Context, pr *github.PRSnapshot) er
180245
return nil
181246
}
182247

248+
// updateClosedPRThread updates Slack threads for a closed or merged PR.
249+
func (c *Coordinator) updateClosedPRThread(ctx context.Context, pr *github.PRSnapshot) error {
250+
prKey := fmt.Sprintf("%s/%s#%d", pr.Owner, pr.Repo, pr.Number)
251+
slog.Debug("updating thread for closed/merged PR",
252+
"pr", prKey,
253+
"state", pr.State)
254+
255+
channels := c.configManager.ChannelsForRepo(pr.Owner, pr.Repo)
256+
if len(channels) == 0 {
257+
slog.Debug("no channels configured for closed PR",
258+
"pr", prKey,
259+
"owner", pr.Owner,
260+
"repo", pr.Repo)
261+
return nil
262+
}
263+
264+
n := 0
265+
for _, ch := range channels {
266+
id := c.slack.ResolveChannelID(ctx, ch)
267+
if id == "" {
268+
slog.Debug("could not resolve channel ID for closed PR thread update",
269+
"channel_name", ch,
270+
"pr", prKey)
271+
continue
272+
}
273+
274+
info, ok := c.stateStore.GetThread(pr.Owner, pr.Repo, pr.Number, id)
275+
if !ok {
276+
slog.Debug("no thread found for closed PR in channel",
277+
"pr", prKey,
278+
"channel", ch,
279+
"channel_id", id)
280+
continue
281+
}
282+
283+
if err := c.updateThreadForClosedPR(ctx, pr, id, info); err != nil {
284+
slog.Warn("failed to update thread for closed PR",
285+
"pr", prKey,
286+
"channel", ch,
287+
"error", err)
288+
continue
289+
}
290+
291+
n++
292+
slog.Info("updated thread for closed/merged PR",
293+
"pr", prKey,
294+
"state", pr.State,
295+
"channel", ch,
296+
"thread_ts", info.ThreadTS)
297+
}
298+
299+
if n == 0 {
300+
return errors.New("no threads found or updated for closed PR")
301+
}
302+
303+
return nil
304+
}
305+
306+
// updateThreadForClosedPR updates a single thread's message to reflect closed/merged state.
307+
func (c *Coordinator) updateThreadForClosedPR(ctx context.Context, pr *github.PRSnapshot, channelID string, info ThreadInfo) error {
308+
var emoji, msg string
309+
switch pr.State {
310+
case "MERGED":
311+
emoji = ":rocket:"
312+
msg = "This PR was merged"
313+
case "CLOSED":
314+
emoji = ":x:"
315+
msg = "This PR was closed without merging"
316+
default:
317+
return fmt.Errorf("unexpected PR state: %s", pr.State)
318+
}
319+
320+
// Replace emoji prefix in message (format: ":emoji: Title • repo#123 by @user")
321+
text := info.MessageText
322+
if i := strings.Index(text, " "); i == -1 {
323+
text = emoji + " " + text
324+
} else {
325+
text = emoji + text[i:]
326+
}
327+
328+
if err := c.slack.UpdateMessage(ctx, channelID, info.ThreadTS, text); err != nil {
329+
return fmt.Errorf("failed to update message: %w", err)
330+
}
331+
332+
// Post follow-up comment (don't fail if this errors - main update succeeded)
333+
if err := c.slack.PostThreadReply(ctx, channelID, info.ThreadTS, msg); err != nil {
334+
slog.Debug("failed to post follow-up comment for closed PR",
335+
"pr", fmt.Sprintf("%s/%s#%d", pr.Owner, pr.Repo, pr.Number),
336+
"error", err)
337+
}
338+
339+
return nil
340+
}
341+
183342
// StartupReconciliation runs once at startup to catch up on any missed notifications.
184343
// This ensures that if the service was down, we still notify about PRs that need attention.
185344
func (c *Coordinator) StartupReconciliation(ctx context.Context) {

‎internal/github/graphql.go‎

Lines changed: 111 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,117 @@ func (c *GraphQLClient) ListOpenPRs(ctx context.Context, org string, updatedSinc
6161
return prs, nil
6262
}
6363

64+
// ListClosedPRs queries all closed/merged PRs for an organization updated in the last N hours.
65+
// This is used to update Slack threads when PRs are closed or merged.
66+
func (c *GraphQLClient) ListClosedPRs(ctx context.Context, org string, updatedSinceHours int) ([]PRSnapshot, error) {
67+
slog.Debug("querying closed/merged PRs via GraphQL",
68+
"org", org,
69+
"updated_since_hours", updatedSinceHours)
70+
71+
since := time.Now().Add(-time.Duration(updatedSinceHours) * time.Hour)
72+
73+
// GraphQL query structure
74+
var query struct {
75+
Search struct {
76+
Nodes []struct {
77+
PullRequest struct {
78+
Number int
79+
Title string
80+
URL string
81+
UpdatedAt time.Time
82+
CreatedAt time.Time
83+
State string
84+
IsDraft bool
85+
Merged bool
86+
Author struct {
87+
Login string
88+
}
89+
Repository struct {
90+
Name string
91+
Owner struct {
92+
Login string
93+
}
94+
}
95+
} `graphql:"... on PullRequest"`
96+
}
97+
PageInfo struct {
98+
EndCursor string
99+
HasNextPage bool
100+
}
101+
} `graphql:"search(query: $searchQuery, type: ISSUE, first: 100, after: $cursor)"`
102+
}
103+
104+
// Build search query: "is:pr is:closed org:X updated:>YYYY-MM-DD"
105+
// This will include both closed-unmerged and merged PRs
106+
searchQuery := fmt.Sprintf("is:pr is:closed org:%s updated:>%s",
107+
org,
108+
since.Format("2006-01-02"))
109+
110+
variables := map[string]any{
111+
"searchQuery": githubv4.String(searchQuery),
112+
"cursor": (*githubv4.String)(nil),
113+
}
114+
115+
var allPRs []PRSnapshot
116+
pageCount := 0
117+
const maxPages = 10
118+
119+
for {
120+
pageCount++
121+
if pageCount > maxPages {
122+
slog.Warn("reached max page limit for closed PR GraphQL query",
123+
"org", org,
124+
"pages", pageCount,
125+
"prs_collected", len(allPRs))
126+
break
127+
}
128+
129+
err := c.client.Query(ctx, &query, variables)
130+
if err != nil {
131+
return nil, fmt.Errorf("GraphQL query failed: %w", err)
132+
}
133+
134+
// Process this page of results
135+
for i := range query.Search.Nodes {
136+
pr := query.Search.Nodes[i].PullRequest
137+
138+
// Determine state: MERGED takes precedence over CLOSED
139+
state := "CLOSED"
140+
if pr.Merged {
141+
state = "MERGED"
142+
}
143+
144+
allPRs = append(allPRs, PRSnapshot{
145+
Owner: pr.Repository.Owner.Login,
146+
Repo: pr.Repository.Name,
147+
Number: pr.Number,
148+
Title: pr.Title,
149+
Author: pr.Author.Login,
150+
URL: pr.URL,
151+
UpdatedAt: pr.UpdatedAt,
152+
CreatedAt: pr.CreatedAt,
153+
State: state,
154+
IsDraft: pr.IsDraft,
155+
})
156+
}
157+
158+
if !query.Search.PageInfo.HasNextPage {
159+
break
160+
}
161+
162+
cursor := githubv4.String(query.Search.PageInfo.EndCursor)
163+
variables["cursor"] = cursor
164+
}
165+
166+
slog.Info("GraphQL query for closed PRs complete",
167+
"org", org,
168+
"total_prs", len(allPRs),
169+
"pages_fetched", pageCount,
170+
"query", searchQuery)
171+
172+
return allPRs, nil
173+
}
174+
64175
// listOpenPRsGraphQL queries using GraphQL for efficiency.
65176
func (c *GraphQLClient) listOpenPRsGraphQL(ctx context.Context, org string, updatedSinceHours int) ([]PRSnapshot, error) {
66177
slog.Debug("querying open PRs via GraphQL",

0 commit comments

Comments
 (0)