Skip to content

Commit c673ab2

Browse files
tstrombergtstromberg
committed
fix SSE streaming
1 parent ab028c7 commit c673ab2

4 files changed

Lines changed: 142 additions & 135 deletions

File tree

‎cmd/prcost/main.go‎

Lines changed: 93 additions & 77 deletions
Original file line numberDiff line numberDiff line change
@@ -296,95 +296,111 @@ func printHumanReadable(breakdown *cost.Breakdown, prURL string) {
296296

297297
// Delay and Future Costs - only show if there are any delay costs
298298
if breakdown.DelayCost > 0 {
299-
// Merge Delay Costs
300-
fmt.Println(" Delay Costs")
301-
fmt.Println(" ───────────")
302-
if breakdown.DelayCostDetail.DeliveryDelayHours > 0 {
303-
if breakdown.DelayCapped {
304-
fmt.Printf(" Delivery %12s %s (capped)\n",
305-
formatCurrency(breakdown.DelayCostDetail.DeliveryDelayCost), formatTimeUnit(breakdown.DelayCostDetail.DeliveryDelayHours))
306-
} else {
307-
fmt.Printf(" Delivery %12s %s\n",
308-
formatCurrency(breakdown.DelayCostDetail.DeliveryDelayCost), formatTimeUnit(breakdown.DelayCostDetail.DeliveryDelayHours))
309-
}
299+
printDelayCosts(breakdown, formatCurrency)
300+
}
301+
302+
// Grand Total
303+
totalHours := breakdown.Author.TotalHours + breakdown.DelayCostDetail.TotalDelayHours
304+
for _, p := range breakdown.Participants {
305+
totalHours += p.TotalHours
306+
}
307+
fmt.Println(" ═══════════════════════════════════════════════════════════════")
308+
fmt.Printf(" Total %12s %s\n",
309+
formatCurrency(breakdown.TotalCost), formatTimeUnit(totalHours))
310+
fmt.Println()
311+
}
312+
313+
// printDelayCosts prints delay and future costs section.
314+
func printDelayCosts(breakdown *cost.Breakdown, formatCurrency func(float64) string) {
315+
// Merge Delay Costs
316+
fmt.Println(" Delay Costs")
317+
fmt.Println(" ───────────")
318+
319+
if breakdown.DelayCostDetail.DeliveryDelayHours > 0 {
320+
cappedSuffix := ""
321+
if breakdown.DelayCapped {
322+
cappedSuffix = " (capped)"
310323
}
324+
fmt.Printf(" Delivery %12s %s%s\n",
325+
formatCurrency(breakdown.DelayCostDetail.DeliveryDelayCost),
326+
formatTimeUnit(breakdown.DelayCostDetail.DeliveryDelayHours),
327+
cappedSuffix)
328+
}
311329

312-
if breakdown.DelayCostDetail.CoordinationHours > 0 {
313-
if breakdown.DelayCapped {
314-
fmt.Printf(" Coordination %12s %s (capped)\n",
315-
formatCurrency(breakdown.DelayCostDetail.CoordinationCost), formatTimeUnit(breakdown.DelayCostDetail.CoordinationHours))
316-
} else {
317-
fmt.Printf(" Coordination %12s %s\n",
318-
formatCurrency(breakdown.DelayCostDetail.CoordinationCost), formatTimeUnit(breakdown.DelayCostDetail.CoordinationHours))
319-
}
330+
if breakdown.DelayCostDetail.CoordinationHours > 0 {
331+
cappedSuffix := ""
332+
if breakdown.DelayCapped {
333+
cappedSuffix = " (capped)"
320334
}
335+
fmt.Printf(" Coordination %12s %s%s\n",
336+
formatCurrency(breakdown.DelayCostDetail.CoordinationCost),
337+
formatTimeUnit(breakdown.DelayCostDetail.CoordinationHours),
338+
cappedSuffix)
339+
}
321340

322-
mergeDelayCost := breakdown.DelayCostDetail.DeliveryDelayCost + breakdown.DelayCostDetail.CoordinationCost
323-
mergeDelayHours := breakdown.DelayCostDetail.DeliveryDelayHours + breakdown.DelayCostDetail.CoordinationHours
324-
fmt.Println(" ────────────")
325-
fmt.Printf(" Subtotal %12s %s\n",
326-
formatCurrency(mergeDelayCost), formatTimeUnit(mergeDelayHours))
327-
fmt.Println()
341+
mergeDelayCost := breakdown.DelayCostDetail.DeliveryDelayCost + breakdown.DelayCostDetail.CoordinationCost
342+
mergeDelayHours := breakdown.DelayCostDetail.DeliveryDelayHours + breakdown.DelayCostDetail.CoordinationHours
343+
fmt.Println(" ────────────")
344+
fmt.Printf(" Subtotal %12s %s\n",
345+
formatCurrency(mergeDelayCost), formatTimeUnit(mergeDelayHours))
346+
fmt.Println()
328347

329-
// Future Costs
330-
hasFutureCosts := breakdown.DelayCostDetail.ReworkPercentage > 0 ||
331-
breakdown.DelayCostDetail.FutureReviewCost > 0 ||
332-
breakdown.DelayCostDetail.FutureMergeCost > 0 ||
333-
breakdown.DelayCostDetail.FutureContextCost > 0
334-
335-
if hasFutureCosts {
336-
fmt.Println(" Future Costs")
337-
fmt.Println(" ────────────")
338-
339-
if breakdown.DelayCostDetail.ReworkPercentage > 0 {
340-
label := fmt.Sprintf("Code Churn (%.0f%% drift)", breakdown.DelayCostDetail.ReworkPercentage)
341-
fmt.Printf(" %-26s%12s %s\n",
342-
label,
343-
formatCurrency(breakdown.DelayCostDetail.CodeChurnCost),
344-
formatTimeUnit(breakdown.DelayCostDetail.CodeChurnHours))
345-
}
348+
// Future Costs
349+
hasFutureCosts := breakdown.DelayCostDetail.ReworkPercentage > 0 ||
350+
breakdown.DelayCostDetail.FutureReviewCost > 0 ||
351+
breakdown.DelayCostDetail.FutureMergeCost > 0 ||
352+
breakdown.DelayCostDetail.FutureContextCost > 0
346353

347-
if breakdown.DelayCostDetail.FutureReviewCost > 0 {
348-
fmt.Printf(" %-26s%12s %s\n",
349-
"Review",
350-
formatCurrency(breakdown.DelayCostDetail.FutureReviewCost), formatTimeUnit(breakdown.DelayCostDetail.FutureReviewHours))
351-
}
354+
if hasFutureCosts {
355+
printFutureCosts(breakdown, formatCurrency)
356+
}
357+
}
352358

353-
if breakdown.DelayCostDetail.FutureMergeCost > 0 {
354-
fmt.Printf(" %-26s%12s %s\n",
355-
"Merge",
356-
formatCurrency(breakdown.DelayCostDetail.FutureMergeCost), formatTimeUnit(breakdown.DelayCostDetail.FutureMergeHours))
357-
}
359+
// printFutureCosts prints future costs subsection.
360+
func printFutureCosts(breakdown *cost.Breakdown, formatCurrency func(float64) string) {
361+
fmt.Println(" Future Costs")
362+
fmt.Println(" ────────────")
363+
364+
if breakdown.DelayCostDetail.ReworkPercentage > 0 {
365+
label := fmt.Sprintf("Code Churn (%.0f%% drift)", breakdown.DelayCostDetail.ReworkPercentage)
366+
fmt.Printf(" %-26s%12s %s\n",
367+
label,
368+
formatCurrency(breakdown.DelayCostDetail.CodeChurnCost),
369+
formatTimeUnit(breakdown.DelayCostDetail.CodeChurnHours))
370+
}
358371

359-
if breakdown.DelayCostDetail.FutureContextCost > 0 {
360-
fmt.Printf(" %-26s%12s %s\n",
361-
"Context Switching",
362-
formatCurrency(breakdown.DelayCostDetail.FutureContextCost), formatTimeUnit(breakdown.DelayCostDetail.FutureContextHours))
363-
}
372+
if breakdown.DelayCostDetail.FutureReviewCost > 0 {
373+
fmt.Printf(" %-26s%12s %s\n",
374+
"Review",
375+
formatCurrency(breakdown.DelayCostDetail.FutureReviewCost),
376+
formatTimeUnit(breakdown.DelayCostDetail.FutureReviewHours))
377+
}
364378

365-
futureCost := breakdown.DelayCostDetail.CodeChurnCost +
366-
breakdown.DelayCostDetail.FutureReviewCost +
367-
breakdown.DelayCostDetail.FutureMergeCost +
368-
breakdown.DelayCostDetail.FutureContextCost
369-
futureHours := breakdown.DelayCostDetail.CodeChurnHours +
370-
breakdown.DelayCostDetail.FutureReviewHours +
371-
breakdown.DelayCostDetail.FutureMergeHours +
372-
breakdown.DelayCostDetail.FutureContextHours
373-
fmt.Println(" ────────────")
374-
fmt.Printf(" Subtotal %12s %s\n",
375-
formatCurrency(futureCost), formatTimeUnit(futureHours))
376-
fmt.Println()
377-
}
379+
if breakdown.DelayCostDetail.FutureMergeCost > 0 {
380+
fmt.Printf(" %-26s%12s %s\n",
381+
"Merge",
382+
formatCurrency(breakdown.DelayCostDetail.FutureMergeCost),
383+
formatTimeUnit(breakdown.DelayCostDetail.FutureMergeHours))
378384
}
379385

380-
// Grand Total
381-
totalHours := breakdown.Author.TotalHours + breakdown.DelayCostDetail.TotalDelayHours
382-
for _, p := range breakdown.Participants {
383-
totalHours += p.TotalHours
386+
if breakdown.DelayCostDetail.FutureContextCost > 0 {
387+
fmt.Printf(" %-26s%12s %s\n",
388+
"Context Switching",
389+
formatCurrency(breakdown.DelayCostDetail.FutureContextCost),
390+
formatTimeUnit(breakdown.DelayCostDetail.FutureContextHours))
384391
}
385-
fmt.Println(" ═══════════════════════════════════════════════════════════════")
386-
fmt.Printf(" Total %12s %s\n",
387-
formatCurrency(breakdown.TotalCost), formatTimeUnit(totalHours))
392+
393+
futureCost := breakdown.DelayCostDetail.CodeChurnCost +
394+
breakdown.DelayCostDetail.FutureReviewCost +
395+
breakdown.DelayCostDetail.FutureMergeCost +
396+
breakdown.DelayCostDetail.FutureContextCost
397+
futureHours := breakdown.DelayCostDetail.CodeChurnHours +
398+
breakdown.DelayCostDetail.FutureReviewHours +
399+
breakdown.DelayCostDetail.FutureMergeHours +
400+
breakdown.DelayCostDetail.FutureContextHours
401+
fmt.Println(" ────────────")
402+
fmt.Printf(" Subtotal %12s %s\n",
403+
formatCurrency(futureCost), formatTimeUnit(futureHours))
388404
fmt.Println()
389405
}
390406

‎internal/server/server.go‎

Lines changed: 20 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -894,17 +894,20 @@ func (s *Server) handleStatic(w http.ResponseWriter, r *http.Request) {
894894
}
895895

896896
// Set content type based on file extension
897-
contentType := "application/octet-stream"
898-
if strings.HasSuffix(path, ".png") {
897+
var contentType string
898+
switch {
899+
case strings.HasSuffix(path, ".png"):
899900
contentType = "image/png"
900-
} else if strings.HasSuffix(path, ".jpg") || strings.HasSuffix(path, ".jpeg") {
901+
case strings.HasSuffix(path, ".jpg"), strings.HasSuffix(path, ".jpeg"):
901902
contentType = "image/jpeg"
902-
} else if strings.HasSuffix(path, ".ico") {
903+
case strings.HasSuffix(path, ".ico"):
903904
contentType = "image/x-icon"
904-
} else if strings.HasSuffix(path, ".css") {
905+
case strings.HasSuffix(path, ".css"):
905906
contentType = "text/css; charset=utf-8"
906-
} else if strings.HasSuffix(path, ".js") {
907+
case strings.HasSuffix(path, ".js"):
907908
contentType = "application/javascript; charset=utf-8"
909+
default:
910+
contentType = "application/octet-stream"
908911
}
909912

910913
w.Header().Set("Content-Type", contentType)
@@ -1560,19 +1563,9 @@ func logSSEError(ctx context.Context, logger *slog.Logger, err error) {
15601563
// processRepoSampleWithProgress processes a repository sample with progress updates via SSE.
15611564
func (s *Server) processRepoSampleWithProgress(ctx context.Context, req *RepoSampleRequest, token string, writer http.ResponseWriter) {
15621565
// Use background context for work to prevent client timeout from canceling operations
1563-
// We'll monitor the request context separately to detect client disconnects
1566+
// The request context (ctx) is only used for SSE writes and logging
15641567
workCtx := context.Background()
15651568

1566-
// Monitor request context for disconnects in background
1567-
disconnected := make(chan struct{})
1568-
go func() {
1569-
<-ctx.Done()
1570-
s.logger.WarnContext(ctx, "[processRepoSampleWithProgress] Client disconnected",
1571-
"error", ctx.Err(),
1572-
"owner", req.Owner,
1573-
"repo", req.Repo)
1574-
close(disconnected)
1575-
}()
15761569
defer func() {
15771570
s.logger.InfoContext(ctx, "[processRepoSampleWithProgress] Stream handler completed",
15781571
"owner", req.Owner,
@@ -1675,18 +1668,9 @@ func (s *Server) processRepoSampleWithProgress(ctx context.Context, req *RepoSam
16751668
// processOrgSampleWithProgress processes an organization sample with progress updates via SSE.
16761669
func (s *Server) processOrgSampleWithProgress(ctx context.Context, req *OrgSampleRequest, token string, writer http.ResponseWriter) {
16771670
// Use background context for work to prevent client timeout from canceling operations
1678-
// We'll monitor the request context separately to detect client disconnects
1671+
// The request context (ctx) is only used for SSE writes and logging
16791672
workCtx := context.Background()
16801673

1681-
// Monitor request context for disconnects in background
1682-
disconnected := make(chan struct{})
1683-
go func() {
1684-
<-ctx.Done()
1685-
s.logger.WarnContext(ctx, "[processOrgSampleWithProgress] Client disconnected",
1686-
"error", ctx.Err(),
1687-
"org", req.Org)
1688-
close(disconnected)
1689-
}()
16901674
defer func() {
16911675
s.logger.InfoContext(ctx, "[processOrgSampleWithProgress] Stream handler completed",
16921676
"org", req.Org)
@@ -1797,6 +1781,7 @@ func (s *Server) processOrgSampleWithProgress(ctx context.Context, req *OrgSampl
17971781
func (s *Server) processPRsInParallel(workCtx, reqCtx context.Context, samples []github.PRSummary, defaultOwner, defaultRepo, token string, cfg cost.Config, writer http.ResponseWriter) []cost.Breakdown {
17981782
var breakdowns []cost.Breakdown
17991783
var mu sync.Mutex
1784+
var sseMu sync.Mutex // Protects SSE writes to prevent corrupted chunked encoding
18001785

18011786
// Use a buffered channel for worker pool pattern
18021787
concurrency := 5 // Process up to 5 PRs concurrently
@@ -1827,13 +1812,15 @@ func (s *Server) processPRsInParallel(workCtx, reqCtx context.Context, samples [
18271812
progress := fmt.Sprintf("%d/%d", index+1, totalSamples)
18281813

18291814
// Send "fetching" update using request context for SSE
1815+
sseMu.Lock()
18301816
logSSEError(reqCtx, s.logger, sendSSE(writer, ProgressUpdate{
18311817
Type: "fetching",
18321818
PR: prSummary.Number,
18331819
Owner: owner,
18341820
Repo: repo,
18351821
Progress: progress,
18361822
}))
1823+
sseMu.Unlock()
18371824

18381825
prURL := fmt.Sprintf("https://github.com/%s/%s/pull/%d", owner, repo, prSummary.Number)
18391826

@@ -1851,6 +1838,7 @@ func (s *Server) processPRsInParallel(workCtx, reqCtx context.Context, samples [
18511838
}
18521839
if err != nil {
18531840
s.logger.WarnContext(reqCtx, "Failed to fetch PR data, skipping", "pr_number", prSummary.Number, "source", s.dataSource, errorKey, err)
1841+
sseMu.Lock()
18541842
logSSEError(reqCtx, s.logger, sendSSE(writer, ProgressUpdate{
18551843
Type: "error",
18561844
PR: prSummary.Number,
@@ -1859,6 +1847,7 @@ func (s *Server) processPRsInParallel(workCtx, reqCtx context.Context, samples [
18591847
Progress: progress,
18601848
Error: fmt.Sprintf("Failed to fetch PR data: %v", err),
18611849
}))
1850+
sseMu.Unlock()
18621851
return
18631852
}
18641853

@@ -1867,13 +1856,15 @@ func (s *Server) processPRsInParallel(workCtx, reqCtx context.Context, samples [
18671856
}
18681857

18691858
// Send "processing" update using request context for SSE
1859+
sseMu.Lock()
18701860
logSSEError(reqCtx, s.logger, sendSSE(writer, ProgressUpdate{
18711861
Type: "processing",
18721862
PR: prSummary.Number,
18731863
Owner: owner,
18741864
Repo: repo,
18751865
Progress: progress,
18761866
}))
1867+
sseMu.Unlock()
18771868

18781869
breakdown := cost.Calculate(prData, cfg)
18791870

@@ -1883,13 +1874,15 @@ func (s *Server) processPRsInParallel(workCtx, reqCtx context.Context, samples [
18831874
mu.Unlock()
18841875

18851876
// Send "complete" update using request context for SSE
1877+
sseMu.Lock()
18861878
logSSEError(reqCtx, s.logger, sendSSE(writer, ProgressUpdate{
18871879
Type: "complete",
18881880
PR: prSummary.Number,
18891881
Owner: owner,
18901882
Repo: repo,
18911883
Progress: progress,
18921884
}))
1885+
sseMu.Unlock()
18931886
}(idx, pr)
18941887
}
18951888

‎pkg/cost/cost.go‎

Lines changed: 14 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -89,20 +89,20 @@ type Config struct {
8989
// DefaultConfig returns reasonable defaults for cost calculation.
9090
func DefaultConfig() Config {
9191
return Config{
92-
AnnualSalary: 249000.0, // Average Staff Software Engineer salary (2025, Glassdoor)
93-
BenefitsMultiplier: 1.3, // 30% benefits overhead
94-
HoursPerYear: 2080.0, // Standard full-time hours
95-
EventDuration: 10 * time.Minute, // 10 minutes per GitHub event
96-
ContextSwitchInDuration: 3 * time.Minute, // 3 min to context switch in (Microsoft Research)
97-
ContextSwitchOutDuration: 16*time.Minute + 33*time.Second, // 16m33s to context switch out (Microsoft Research)
98-
SessionGapThreshold: 20 * time.Minute, // Events within 20 min are same session
99-
DeliveryDelayFactor: 0.15, // 15% opportunity cost
100-
CoordinationFactor: 0.05, // 5% mental overhead
101-
MaxDelayAfterLastEvent: 14 * 24 * time.Hour, // 14 days (2 weeks) after last event
102-
MaxProjectDelay: 90 * 24 * time.Hour, // 90 days absolute max
103-
MaxCodeDrift: 90 * 24 * time.Hour, // 90 days
104-
ReviewInspectionRate: 275.0, // 275 LOC/hour (average of optimal 150-400 range)
105-
ModificationCostFactor: 0.4, // Modified code costs 40% of new code
92+
AnnualSalary: 249000.0, // Average Staff Software Engineer salary (2025, Glassdoor)
93+
BenefitsMultiplier: 1.3, // 30% benefits overhead
94+
HoursPerYear: 2080.0, // Standard full-time hours
95+
EventDuration: 10 * time.Minute, // 10 minutes per GitHub event
96+
ContextSwitchInDuration: 3 * time.Minute, // 3 min to context switch in (Microsoft Research)
97+
ContextSwitchOutDuration: 16*time.Minute + 33*time.Second, // 16m33s to context switch out (Microsoft Research)
98+
SessionGapThreshold: 20 * time.Minute, // Events within 20 min are same session
99+
DeliveryDelayFactor: 0.15, // 15% opportunity cost
100+
CoordinationFactor: 0.05, // 5% mental overhead
101+
MaxDelayAfterLastEvent: 14 * 24 * time.Hour, // 14 days (2 weeks) after last event
102+
MaxProjectDelay: 90 * 24 * time.Hour, // 90 days absolute max
103+
MaxCodeDrift: 90 * 24 * time.Hour, // 90 days
104+
ReviewInspectionRate: 275.0, // 275 LOC/hour (average of optimal 150-400 range)
105+
ModificationCostFactor: 0.4, // Modified code costs 40% of new code
106106
COCOMO: cocomo.DefaultConfig(),
107107
}
108108
}

0 commit comments

Comments
 (0)