feat: Wire MQ lifecycle events to gt feed display (gt-lak31)
- Add MQ event types and logging in mrqueue/events.go - Have refinery emit merge_started, merged, merge_failed, merge_skipped events - Create MQEventSource to read from mq_events.jsonl - Add MultiSource to combine events from bd activity and MQ events - Add color coding: green for merged, red for failed - Update feed help with MQ event symbols Events are stored in .beads/mq_events.jsonl and displayed in the feed TUI with appropriate symbols and colors. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Steve Yegge
@@ -0,0 +1,189 @@
|
||||
package feed
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/steveyegge/gastown/internal/mrqueue"
|
||||
)
|
||||
|
||||
// MQEventSource reads MQ lifecycle events from mq_events.jsonl
|
||||
type MQEventSource struct {
|
||||
file *os.File
|
||||
events chan Event
|
||||
cancel context.CancelFunc
|
||||
logPath string
|
||||
}
|
||||
|
||||
// NewMQEventSource creates a source that tails MQ events from a beads directory.
|
||||
func NewMQEventSource(beadsDir string) (*MQEventSource, error) {
|
||||
logPath := filepath.Join(beadsDir, "mq_events.jsonl")
|
||||
|
||||
// Create file if it doesn't exist
|
||||
if _, err := os.Stat(logPath); os.IsNotExist(err) {
|
||||
// Ensure directory exists
|
||||
if err := os.MkdirAll(filepath.Dir(logPath), 0755); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// Create empty file
|
||||
f, err := os.Create(logPath)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
f.Close()
|
||||
}
|
||||
|
||||
file, err := os.Open(logPath)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
source := &MQEventSource{
|
||||
file: file,
|
||||
events: make(chan Event, 100),
|
||||
cancel: cancel,
|
||||
logPath: logPath,
|
||||
}
|
||||
|
||||
go source.tail(ctx)
|
||||
|
||||
return source, nil
|
||||
}
|
||||
|
||||
// NewMQEventSourceFromWorkDir creates an MQ event source by finding the beads directory.
|
||||
func NewMQEventSourceFromWorkDir(workDir string) (*MQEventSource, error) {
|
||||
beadsDir, err := FindBeadsDir(workDir)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return NewMQEventSource(beadsDir)
|
||||
}
|
||||
|
||||
// tail follows the MQ event log file and sends events.
|
||||
func (s *MQEventSource) tail(ctx context.Context) {
|
||||
defer close(s.events)
|
||||
|
||||
// Seek to end for live tailing
|
||||
s.file.Seek(0, 2)
|
||||
|
||||
scanner := bufio.NewScanner(s.file)
|
||||
ticker := time.NewTicker(100 * time.Millisecond)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
for scanner.Scan() {
|
||||
line := scanner.Text()
|
||||
if event := parseMQEventLine(line); event != nil {
|
||||
select {
|
||||
case s.events <- *event:
|
||||
default:
|
||||
// Drop event if channel full
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Events returns the event channel.
|
||||
func (s *MQEventSource) Events() <-chan Event {
|
||||
return s.events
|
||||
}
|
||||
|
||||
// Close stops the source.
|
||||
func (s *MQEventSource) Close() error {
|
||||
s.cancel()
|
||||
return s.file.Close()
|
||||
}
|
||||
|
||||
// parseMQEventLine parses a line from mq_events.jsonl into a feed Event.
|
||||
func parseMQEventLine(line string) *Event {
|
||||
if strings.TrimSpace(line) == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
var mqEvent mrqueue.Event
|
||||
if err := json.Unmarshal([]byte(line), &mqEvent); err != nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Convert MQ event to feed Event
|
||||
feedType := mapMQEventType(mqEvent.Type)
|
||||
message := formatMQEventMessage(mqEvent)
|
||||
|
||||
return &Event{
|
||||
Time: mqEvent.Timestamp,
|
||||
Type: feedType,
|
||||
Actor: "refinery",
|
||||
Target: mqEvent.MRID,
|
||||
Message: message,
|
||||
Rig: mqEvent.Rig,
|
||||
Role: "refinery",
|
||||
Raw: line,
|
||||
}
|
||||
}
|
||||
|
||||
// mapMQEventType maps MQ event types to feed event types.
|
||||
func mapMQEventType(mqType mrqueue.EventType) string {
|
||||
switch mqType {
|
||||
case mrqueue.EventMergeStarted:
|
||||
return "merge_started"
|
||||
case mrqueue.EventMerged:
|
||||
return "merged"
|
||||
case mrqueue.EventMergeFailed:
|
||||
return "merge_failed"
|
||||
case mrqueue.EventMergeSkipped:
|
||||
return "merge_skipped"
|
||||
default:
|
||||
return string(mqType)
|
||||
}
|
||||
}
|
||||
|
||||
// formatMQEventMessage creates a human-readable message for an MQ event.
|
||||
func formatMQEventMessage(e mrqueue.Event) string {
|
||||
branchInfo := e.Branch
|
||||
if e.Target != "" {
|
||||
branchInfo += " -> " + e.Target
|
||||
}
|
||||
|
||||
switch e.Type {
|
||||
case mrqueue.EventMergeStarted:
|
||||
return "Merge started: " + branchInfo
|
||||
case mrqueue.EventMerged:
|
||||
msg := "Merged: " + branchInfo
|
||||
if e.MergeCommit != "" {
|
||||
// Show short commit SHA
|
||||
sha := e.MergeCommit
|
||||
if len(sha) > 8 {
|
||||
sha = sha[:8]
|
||||
}
|
||||
msg += " (" + sha + ")"
|
||||
}
|
||||
return msg
|
||||
case mrqueue.EventMergeFailed:
|
||||
msg := "Merge failed: " + branchInfo
|
||||
if e.Reason != "" {
|
||||
msg += " - " + e.Reason
|
||||
}
|
||||
return msg
|
||||
case mrqueue.EventMergeSkipped:
|
||||
msg := "Merge skipped: " + branchInfo
|
||||
if e.Reason != "" {
|
||||
msg += " - " + e.Reason
|
||||
}
|
||||
return msg
|
||||
default:
|
||||
return string(e.Type) + ": " + branchInfo
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,144 @@
|
||||
package feed
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/steveyegge/gastown/internal/mrqueue"
|
||||
)
|
||||
|
||||
func TestParseMQEventLine(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
event mrqueue.Event
|
||||
wantType string
|
||||
wantTarget string
|
||||
wantContains string // Substring in message
|
||||
}{
|
||||
{
|
||||
name: "merge_started",
|
||||
event: mrqueue.Event{
|
||||
Timestamp: time.Now(),
|
||||
Type: mrqueue.EventMergeStarted,
|
||||
MRID: "mr-123",
|
||||
Branch: "polecat/nux",
|
||||
Target: "main",
|
||||
Worker: "nux",
|
||||
Rig: "gastown",
|
||||
},
|
||||
wantType: "merge_started",
|
||||
wantTarget: "mr-123",
|
||||
wantContains: "Merge started",
|
||||
},
|
||||
{
|
||||
name: "merged",
|
||||
event: mrqueue.Event{
|
||||
Timestamp: time.Now(),
|
||||
Type: mrqueue.EventMerged,
|
||||
MRID: "mr-456",
|
||||
Branch: "polecat/toast",
|
||||
Target: "main",
|
||||
Worker: "toast",
|
||||
Rig: "gastown",
|
||||
MergeCommit: "abc123def456789",
|
||||
},
|
||||
wantType: "merged",
|
||||
wantTarget: "mr-456",
|
||||
wantContains: "abc123de", // Short SHA
|
||||
},
|
||||
{
|
||||
name: "merge_failed",
|
||||
event: mrqueue.Event{
|
||||
Timestamp: time.Now(),
|
||||
Type: mrqueue.EventMergeFailed,
|
||||
MRID: "mr-789",
|
||||
Branch: "polecat/capable",
|
||||
Target: "main",
|
||||
Worker: "capable",
|
||||
Rig: "gastown",
|
||||
Reason: "conflict in main.go",
|
||||
},
|
||||
wantType: "merge_failed",
|
||||
wantTarget: "mr-789",
|
||||
wantContains: "conflict in main.go",
|
||||
},
|
||||
{
|
||||
name: "merge_skipped",
|
||||
event: mrqueue.Event{
|
||||
Timestamp: time.Now(),
|
||||
Type: mrqueue.EventMergeSkipped,
|
||||
MRID: "mr-999",
|
||||
Branch: "polecat/skip",
|
||||
Target: "main",
|
||||
Reason: "already merged",
|
||||
},
|
||||
wantType: "merge_skipped",
|
||||
wantTarget: "mr-999",
|
||||
wantContains: "already merged",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
// Marshal to JSON line
|
||||
data, err := json.Marshal(tt.event)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to marshal event: %v", err)
|
||||
}
|
||||
|
||||
// Parse the line
|
||||
result := parseMQEventLine(string(data))
|
||||
if result == nil {
|
||||
t.Fatal("parseMQEventLine returned nil")
|
||||
}
|
||||
|
||||
if result.Type != tt.wantType {
|
||||
t.Errorf("Type = %q, want %q", result.Type, tt.wantType)
|
||||
}
|
||||
|
||||
if result.Target != tt.wantTarget {
|
||||
t.Errorf("Target = %q, want %q", result.Target, tt.wantTarget)
|
||||
}
|
||||
|
||||
if tt.wantContains != "" && !contains(result.Message, tt.wantContains) {
|
||||
t.Errorf("Message = %q, want to contain %q", result.Message, tt.wantContains)
|
||||
}
|
||||
|
||||
// Actor should be refinery
|
||||
if result.Actor != "refinery" {
|
||||
t.Errorf("Actor = %q, want %q", result.Actor, "refinery")
|
||||
}
|
||||
|
||||
if result.Role != "refinery" {
|
||||
t.Errorf("Role = %q, want %q", result.Role, "refinery")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseMQEventLineEmpty(t *testing.T) {
|
||||
result := parseMQEventLine("")
|
||||
if result != nil {
|
||||
t.Error("Expected nil for empty line")
|
||||
}
|
||||
|
||||
result = parseMQEventLine(" ")
|
||||
if result != nil {
|
||||
t.Error("Expected nil for whitespace-only line")
|
||||
}
|
||||
|
||||
result = parseMQEventLine("not valid json")
|
||||
if result != nil {
|
||||
t.Error("Expected nil for invalid JSON")
|
||||
}
|
||||
}
|
||||
|
||||
func contains(s, substr string) bool {
|
||||
for i := 0; i <= len(s)-len(substr); i++ {
|
||||
if s[i:i+len(substr)] == substr {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -0,0 +1,80 @@
|
||||
package feed
|
||||
|
||||
import (
|
||||
"sync"
|
||||
)
|
||||
|
||||
// MultiSource combines events from multiple EventSources into a single stream.
|
||||
type MultiSource struct {
|
||||
sources []EventSource
|
||||
events chan Event
|
||||
done chan struct{}
|
||||
wg sync.WaitGroup
|
||||
}
|
||||
|
||||
// NewMultiSource creates a new multi-source that combines events from all given sources.
|
||||
func NewMultiSource(sources ...EventSource) *MultiSource {
|
||||
m := &MultiSource{
|
||||
sources: sources,
|
||||
events: make(chan Event, 100),
|
||||
done: make(chan struct{}),
|
||||
}
|
||||
|
||||
// Start a goroutine for each source to forward events
|
||||
for _, src := range sources {
|
||||
if src == nil {
|
||||
continue
|
||||
}
|
||||
m.wg.Add(1)
|
||||
go m.forwardEvents(src)
|
||||
}
|
||||
|
||||
// Close events channel when all sources are done
|
||||
go func() {
|
||||
m.wg.Wait()
|
||||
close(m.events)
|
||||
}()
|
||||
|
||||
return m
|
||||
}
|
||||
|
||||
// forwardEvents reads from a source and forwards to the combined channel.
|
||||
func (m *MultiSource) forwardEvents(src EventSource) {
|
||||
defer m.wg.Done()
|
||||
|
||||
srcEvents := src.Events()
|
||||
for {
|
||||
select {
|
||||
case event, ok := <-srcEvents:
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
select {
|
||||
case m.events <- event:
|
||||
case <-m.done:
|
||||
return
|
||||
}
|
||||
case <-m.done:
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Events returns the combined event channel.
|
||||
func (m *MultiSource) Events() <-chan Event {
|
||||
return m.events
|
||||
}
|
||||
|
||||
// Close stops all sources.
|
||||
func (m *MultiSource) Close() error {
|
||||
close(m.done)
|
||||
var lastErr error
|
||||
for _, src := range m.sources {
|
||||
if src != nil {
|
||||
if err := src.Close(); err != nil {
|
||||
lastErr = err
|
||||
}
|
||||
}
|
||||
}
|
||||
return lastErr
|
||||
}
|
||||
@@ -106,13 +106,32 @@ var (
|
||||
"deacon": "🔔",
|
||||
}
|
||||
|
||||
// MQ event styles
|
||||
EventMergeStartedStyle = lipgloss.NewStyle().
|
||||
Foreground(colorPrimary)
|
||||
|
||||
EventMergedStyle = lipgloss.NewStyle().
|
||||
Foreground(colorSuccess).
|
||||
Bold(true)
|
||||
|
||||
EventMergeFailedStyle = lipgloss.NewStyle().
|
||||
Foreground(colorError).
|
||||
Bold(true)
|
||||
|
||||
EventMergeSkippedStyle = lipgloss.NewStyle().
|
||||
Foreground(colorWarning)
|
||||
|
||||
// Event symbols
|
||||
EventSymbols = map[string]string{
|
||||
"create": "+",
|
||||
"update": "→",
|
||||
"complete": "✓",
|
||||
"fail": "✗",
|
||||
"delete": "⊘",
|
||||
"pin": "📌",
|
||||
"create": "+",
|
||||
"update": "→",
|
||||
"complete": "✓",
|
||||
"fail": "✗",
|
||||
"delete": "⊘",
|
||||
"pin": "📌",
|
||||
"merge_started": "⚙",
|
||||
"merged": "✓",
|
||||
"merge_failed": "✗",
|
||||
"merge_skipped": "⊘",
|
||||
}
|
||||
)
|
||||
|
||||
@@ -255,6 +255,14 @@ func (m *Model) renderEvent(e Event) string {
|
||||
symbolStyle = EventFailStyle
|
||||
case "delete":
|
||||
symbolStyle = EventDeleteStyle
|
||||
case "merge_started":
|
||||
symbolStyle = EventMergeStartedStyle
|
||||
case "merged":
|
||||
symbolStyle = EventMergedStyle
|
||||
case "merge_failed":
|
||||
symbolStyle = EventMergeFailedStyle
|
||||
case "merge_skipped":
|
||||
symbolStyle = EventMergeSkippedStyle
|
||||
default:
|
||||
symbolStyle = EventUpdateStyle
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user