diff --git a/internal/cost/cost_tracker.go b/internal/cost/cost_tracker.go new file mode 100644 index 0000000..4161c37 --- /dev/null +++ b/internal/cost/cost_tracker.go @@ -0,0 +1,246 @@ +package cost + +import ( + "fmt" + "sync" + "time" +) + +// CostEntry represents a tracked cost +type CostEntry struct { + ID string + Type string // llm, git, compute + WorkflowID string + TaskID string + Amount float64 + Timestamp time.Time + Metadata map[string]interface{} +} + +// CostTracker tracks and analyzes workflow costs +type CostTracker struct { + mu sync.RWMutex + entries []*CostEntry + rates map[string]float64 +} + +// NewCostTracker creates a new cost tracker +func NewCostTracker() *CostTracker { + return &CostTracker{ + entries: make([]*CostEntry, 0), + rates: map[string]float64{ + "llm_token": 0.0001, // $0.0001 per token + "git_push": 0.0, // Free + "compute_hour": 0.5, // $0.5 per hour + }, + } +} + +// TrackLLMCost tracks LLM API costs +func (ct *CostTracker) TrackLLMCost(workflowID, taskID string, tokens int) { + cost := float64(tokens) * ct.rates["llm_token"] + ct.mu.Lock() + defer ct.mu.Unlock() + + entry := &CostEntry{ + ID: fmt.Sprintf("llm-%d", len(ct.entries)), + Type: "llm", + WorkflowID: workflowID, + TaskID: taskID, + Amount: cost, + Timestamp: time.Now(), + Metadata: map[string]interface{}{ + "tokens": tokens, + }, + } + + ct.entries = append(ct.entries, entry) +} + +// TrackGitCost tracks git operation costs +func (ct *CostTracker) TrackGitCost(workflowID string, operations int) { + ct.mu.Lock() + defer ct.mu.Unlock() + + entry := &CostEntry{ + ID: fmt.Sprintf("git-%d", len(ct.entries)), + Type: "git", + WorkflowID: workflowID, + Amount: 0, + Timestamp: time.Now(), + Metadata: map[string]interface{}{ + "operations": operations, + }, + } + + ct.entries = append(ct.entries, entry) +} + +// TrackComputeCost tracks compute resource costs (duration in milliseconds) +func (ct *CostTracker) TrackComputeCost(workflowID, taskID string, durationMs float64) { + // Convert milliseconds to hours + durationHours := durationMs / (1000.0 * 3600.0) + cost := durationHours * ct.rates["compute_hour"] + ct.mu.Lock() + defer ct.mu.Unlock() + + entry := &CostEntry{ + ID: fmt.Sprintf("compute-%d", len(ct.entries)), + Type: "compute", + WorkflowID: workflowID, + TaskID: taskID, + Amount: cost, + Timestamp: time.Now(), + Metadata: map[string]interface{}{ + "duration_ms": durationMs, + }, + } + + ct.entries = append(ct.entries, entry) +} + +// GetTotalCost returns total cost for all workflows +func (ct *CostTracker) GetTotalCost() float64 { + ct.mu.RLock() + defer ct.mu.RUnlock() + + total := 0.0 + for _, entry := range ct.entries { + total += entry.Amount + } + + return total +} + +// GetWorkflowCost returns total cost for a specific workflow +func (ct *CostTracker) GetWorkflowCost(workflowID string) float64 { + ct.mu.RLock() + defer ct.mu.RUnlock() + + total := 0.0 + for _, entry := range ct.entries { + if entry.WorkflowID == workflowID { + total += entry.Amount + } + } + + return total +} + +// GetCostByType returns total cost by type +func (ct *CostTracker) GetCostByType(costType string) float64 { + ct.mu.RLock() + defer ct.mu.RUnlock() + + total := 0.0 + for _, entry := range ct.entries { + if entry.Type == costType { + total += entry.Amount + } + } + + return total +} + +// GetAverageCostPerTask returns average cost per task +func (ct *CostTracker) GetAverageCostPerTask(workflowID string) float64 { + ct.mu.RLock() + defer ct.mu.RUnlock() + + total := 0.0 + count := 0 + + for _, entry := range ct.entries { + if entry.WorkflowID == workflowID { + total += entry.Amount + count++ + } + } + + if count == 0 { + return 0 + } + + return total / float64(count) +} + +// GetOptimizationSuggestions returns cost optimization recommendations +func (ct *CostTracker) GetOptimizationSuggestions(workflowID string) []string { + suggestions := make([]string, 0) + + ct.mu.RLock() + defer ct.mu.RUnlock() + + llmCost := 0.0 + computeCost := 0.0 + + for _, entry := range ct.entries { + if entry.WorkflowID == workflowID { + if entry.Type == "llm" { + llmCost += entry.Amount + } else if entry.Type == "compute" { + computeCost += entry.Amount + } + } + } + + if llmCost > computeCost*2 { + suggestions = append(suggestions, "Consider caching LLM results to reduce API calls") + } + + if computeCost > llmCost*2 { + suggestions = append(suggestions, "Consider parallelizing compute tasks") + } + + return suggestions +} + +// GetEntries returns all cost entries +func (ct *CostTracker) GetEntries() []*CostEntry { + ct.mu.RLock() + defer ct.mu.RUnlock() + + result := make([]*CostEntry, len(ct.entries)) + copy(result, ct.entries) + + return result +} + +// GetEntriesForWorkflow returns cost entries for a workflow +func (ct *CostTracker) GetEntriesForWorkflow(workflowID string) []*CostEntry { + ct.mu.RLock() + defer ct.mu.RUnlock() + + result := make([]*CostEntry, 0) + for _, entry := range ct.entries { + if entry.WorkflowID == workflowID { + result = append(result, entry) + } + } + + return result +} + +// SetRate sets the cost rate for a type +func (ct *CostTracker) SetRate(costType string, rate float64) { + ct.mu.Lock() + defer ct.mu.Unlock() + + ct.rates[costType] = rate +} + +// GetRate gets the cost rate for a type +func (ct *CostTracker) GetRate(costType string) float64 { + ct.mu.RLock() + defer ct.mu.RUnlock() + + return ct.rates[costType] +} + +// Clear clears all cost entries +func (ct *CostTracker) Clear() { + ct.mu.Lock() + defer ct.mu.Unlock() + + ct.entries = make([]*CostEntry, 0) +} diff --git a/internal/cost/cost_tracker_test.go b/internal/cost/cost_tracker_test.go new file mode 100644 index 0000000..7b798c9 --- /dev/null +++ b/internal/cost/cost_tracker_test.go @@ -0,0 +1,152 @@ +package cost + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestTrackLLMCost(t *testing.T) { + tracker := NewCostTracker() + tracker.TrackLLMCost("wf-1", "task-1", 1000) + + entries := tracker.GetEntries() + assert.Equal(t, 1, len(entries)) + assert.Equal(t, "llm", entries[0].Type) + assert.Equal(t, 0.1, entries[0].Amount) // 1000 tokens * 0.0001 +} + +func TestTrackGitCost(t *testing.T) { + tracker := NewCostTracker() + tracker.TrackGitCost("wf-1", 5) + + entries := tracker.GetEntries() + assert.Equal(t, 1, len(entries)) + assert.Equal(t, "git", entries[0].Type) +} + +func TestTrackComputeCost(t *testing.T) { + tracker := NewCostTracker() + tracker.TrackComputeCost("wf-1", "task-1", 3600000) // 1 hour in ms + + entries := tracker.GetEntries() + assert.Equal(t, 1, len(entries)) + assert.Equal(t, "compute", entries[0].Type) + assert.Equal(t, 0.5, entries[0].Amount) // 1 hour * $0.5/hour +} + +func TestGetTotalCost(t *testing.T) { + tracker := NewCostTracker() + + tracker.TrackLLMCost("wf-1", "task-1", 1000) + tracker.TrackComputeCost("wf-1", "task-1", 3600000) + + total := tracker.GetTotalCost() + assert.Equal(t, 0.6, total) // 0.1 + 0.5 +} + +func TestGetWorkflowCost(t *testing.T) { + tracker := NewCostTracker() + + tracker.TrackLLMCost("wf-1", "task-1", 1000) + tracker.TrackLLMCost("wf-2", "task-1", 2000) + + cost := tracker.GetWorkflowCost("wf-1") + assert.Equal(t, 0.1, cost) + + cost = tracker.GetWorkflowCost("wf-2") + assert.Equal(t, 0.2, cost) +} + +func TestGetCostByType(t *testing.T) { + tracker := NewCostTracker() + + tracker.TrackLLMCost("wf-1", "task-1", 1000) + tracker.TrackLLMCost("wf-1", "task-2", 1000) + tracker.TrackComputeCost("wf-1", "task-3", 3600000) + + llmCost := tracker.GetCostByType("llm") + assert.Equal(t, 0.2, llmCost) + + computeCost := tracker.GetCostByType("compute") + assert.Equal(t, 0.5, computeCost) +} + +func TestGetAverageCostPerTask(t *testing.T) { + tracker := NewCostTracker() + + tracker.TrackLLMCost("wf-1", "task-1", 1000) + tracker.TrackLLMCost("wf-1", "task-2", 1000) + + avg := tracker.GetAverageCostPerTask("wf-1") + assert.Equal(t, 0.1, avg) +} + +func TestGetOptimizationSuggestions(t *testing.T) { + tracker := NewCostTracker() + + // High LLM cost + tracker.TrackLLMCost("wf-1", "task-1", 10000) + tracker.TrackLLMCost("wf-1", "task-2", 10000) + tracker.TrackComputeCost("wf-1", "task-3", 360000) // 0.1 seconds + + suggestions := tracker.GetOptimizationSuggestions("wf-1") + // Just verify it returns without error - suggestions depend on cost ratios + assert.NotNil(t, suggestions) +} + +func TestGetEntriesForWorkflow(t *testing.T) { + tracker := NewCostTracker() + + tracker.TrackLLMCost("wf-1", "task-1", 1000) + tracker.TrackLLMCost("wf-2", "task-1", 1000) + + entries := tracker.GetEntriesForWorkflow("wf-1") + assert.Equal(t, 1, len(entries)) +} + +func TestSetAndGetRate(t *testing.T) { + tracker := NewCostTracker() + + tracker.SetRate("custom", 0.5) + rate := tracker.GetRate("custom") + + assert.Equal(t, 0.5, rate) +} + +func TestClear(t *testing.T) { + tracker := NewCostTracker() + + tracker.TrackLLMCost("wf-1", "task-1", 1000) + tracker.Clear() + + entries := tracker.GetEntries() + assert.Equal(t, 0, len(entries)) +} + +func TestMultipleCosts(t *testing.T) { + tracker := NewCostTracker() + + tracker.TrackLLMCost("wf-1", "task-1", 1000) + tracker.TrackGitCost("wf-1", 5) + tracker.TrackComputeCost("wf-1", "task-1", 1800000) // 30 min + + total := tracker.GetTotalCost() + assert.True(t, total > 0.2) +} + +func TestZeroCost(t *testing.T) { + tracker := NewCostTracker() + + cost := tracker.GetWorkflowCost("nonexistent") + assert.Equal(t, 0.0, cost) +} + +func BenchmarkTrackLLMCost(b *testing.B) { + tracker := NewCostTracker() + + b.ResetTimer() + for i := 0; i < b.N; i++ { + tracker.TrackLLMCost("wf-1", "task-1", 1000) + } +} diff --git a/internal/dashboard/metrics_aggregator.go b/internal/dashboard/metrics_aggregator.go new file mode 100644 index 0000000..6927b57 --- /dev/null +++ b/internal/dashboard/metrics_aggregator.go @@ -0,0 +1,203 @@ +package dashboard + +import ( + "fmt" + "sort" + "sync" + "time" +) + +// MetricSnapshot represents a point-in-time metric value +type MetricSnapshot struct { + Timestamp time.Time + Value float64 + Name string +} + +// MetricsAggregator aggregates Prometheus metrics for dashboard display +type MetricsAggregator struct { + mu sync.RWMutex + metrics map[string][]MetricSnapshot + ttl time.Duration + maxSize int +} + +// NewMetricsAggregator creates a new metrics aggregator +func NewMetricsAggregator(ttl time.Duration, maxSize int) *MetricsAggregator { + return &MetricsAggregator{ + metrics: make(map[string][]MetricSnapshot), + ttl: ttl, + maxSize: maxSize, + } +} + +// Record records a metric value +func (ma *MetricsAggregator) Record(name string, value float64) { + ma.mu.Lock() + defer ma.mu.Unlock() + + snapshot := MetricSnapshot{ + Timestamp: time.Now(), + Value: value, + Name: name, + } + + ma.metrics[name] = append(ma.metrics[name], snapshot) + + // Trim old entries + if len(ma.metrics[name]) > ma.maxSize { + ma.metrics[name] = ma.metrics[name][1:] + } +} + +// GetTimeSeries retrieves metric time series +func (ma *MetricsAggregator) GetTimeSeries(name string) []MetricSnapshot { + ma.mu.RLock() + defer ma.mu.RUnlock() + + snapshots, exists := ma.metrics[name] + if !exists { + return []MetricSnapshot{} + } + + result := make([]MetricSnapshot, len(snapshots)) + copy(result, snapshots) + return result +} + +// GetPercentile calculates percentile for a metric +func (ma *MetricsAggregator) GetPercentile(name string, percentile float64) (float64, error) { + ma.mu.RLock() + defer ma.mu.RUnlock() + + snapshots, exists := ma.metrics[name] + if !exists || len(snapshots) == 0 { + return 0, fmt.Errorf("metric not found: %s", name) + } + + values := make([]float64, len(snapshots)) + for i, s := range snapshots { + values[i] = s.Value + } + + sort.Float64s(values) + + index := int(float64(len(values)) * percentile / 100) + if index >= len(values) { + index = len(values) - 1 + } + + return values[index], nil +} + +// GetAverage calculates average for a metric +func (ma *MetricsAggregator) GetAverage(name string) (float64, error) { + ma.mu.RLock() + defer ma.mu.RUnlock() + + snapshots, exists := ma.metrics[name] + if !exists || len(snapshots) == 0 { + return 0, fmt.Errorf("metric not found: %s", name) + } + + sum := 0.0 + for _, s := range snapshots { + sum += s.Value + } + + return sum / float64(len(snapshots)), nil +} + +// GetMax returns maximum value for a metric +func (ma *MetricsAggregator) GetMax(name string) (float64, error) { + ma.mu.RLock() + defer ma.mu.RUnlock() + + snapshots, exists := ma.metrics[name] + if !exists || len(snapshots) == 0 { + return 0, fmt.Errorf("metric not found: %s", name) + } + + max := snapshots[0].Value + for _, s := range snapshots { + if s.Value > max { + max = s.Value + } + } + + return max, nil +} + +// GetMin returns minimum value for a metric +func (ma *MetricsAggregator) GetMin(name string) (float64, error) { + ma.mu.RLock() + defer ma.mu.RUnlock() + + snapshots, exists := ma.metrics[name] + if !exists || len(snapshots) == 0 { + return 0, fmt.Errorf("metric not found: %s", name) + } + + min := snapshots[0].Value + for _, s := range snapshots { + if s.Value < min { + min = s.Value + } + } + + return min, nil +} + +// GetMetricNames returns all recorded metric names +func (ma *MetricsAggregator) GetMetricNames() []string { + ma.mu.RLock() + defer ma.mu.RUnlock() + + names := make([]string, 0, len(ma.metrics)) + for name := range ma.metrics { + names = append(names, name) + } + + return names +} + +// GetLatest returns the latest snapshot for a metric +func (ma *MetricsAggregator) GetLatest(name string) (MetricSnapshot, error) { + ma.mu.RLock() + defer ma.mu.RUnlock() + + snapshots, exists := ma.metrics[name] + if !exists || len(snapshots) == 0 { + return MetricSnapshot{}, fmt.Errorf("metric not found: %s", name) + } + + return snapshots[len(snapshots)-1], nil +} + +// Clear clears all metrics +func (ma *MetricsAggregator) Clear() { + ma.mu.Lock() + defer ma.mu.Unlock() + + ma.metrics = make(map[string][]MetricSnapshot) +} + +// GetCountInRange returns count of metrics within a time range +func (ma *MetricsAggregator) GetCountInRange(name string, start, end time.Time) (int, error) { + ma.mu.RLock() + defer ma.mu.RUnlock() + + snapshots, exists := ma.metrics[name] + if !exists { + return 0, fmt.Errorf("metric not found: %s", name) + } + + count := 0 + for _, s := range snapshots { + if s.Timestamp.After(start) && s.Timestamp.Before(end) { + count++ + } + } + + return count, nil +} diff --git a/internal/dashboard/metrics_aggregator_test.go b/internal/dashboard/metrics_aggregator_test.go new file mode 100644 index 0000000..e240385 --- /dev/null +++ b/internal/dashboard/metrics_aggregator_test.go @@ -0,0 +1,155 @@ +package dashboard + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +func TestRecord(t *testing.T) { + agg := NewMetricsAggregator(1*time.Hour, 100) + agg.Record("request_latency", 150.5) + + names := agg.GetMetricNames() + assert.Equal(t, 1, len(names)) + assert.Equal(t, "request_latency", names[0]) +} + +func TestGetTimeSeries(t *testing.T) { + agg := NewMetricsAggregator(1*time.Hour, 100) + + agg.Record("latency", 100) + agg.Record("latency", 200) + agg.Record("latency", 150) + + series := agg.GetTimeSeries("latency") + assert.Equal(t, 3, len(series)) + assert.Equal(t, 100.0, series[0].Value) + assert.Equal(t, 200.0, series[1].Value) + assert.Equal(t, 150.0, series[2].Value) +} + +func TestGetPercentile(t *testing.T) { + agg := NewMetricsAggregator(1*time.Hour, 100) + + for i := 1; i <= 100; i++ { + agg.Record("latency", float64(i)) + } + + p50, _ := agg.GetPercentile("latency", 50) + p95, _ := agg.GetPercentile("latency", 95) + p99, _ := agg.GetPercentile("latency", 99) + + assert.True(t, p50 > 40 && p50 < 60) + assert.True(t, p95 > 90) + assert.True(t, p99 > 95) +} + +func TestGetAverage(t *testing.T) { + agg := NewMetricsAggregator(1*time.Hour, 100) + + agg.Record("latency", 100) + agg.Record("latency", 200) + agg.Record("latency", 300) + + avg, _ := agg.GetAverage("latency") + assert.Equal(t, 200.0, avg) +} + +func TestGetMax(t *testing.T) { + agg := NewMetricsAggregator(1*time.Hour, 100) + + agg.Record("latency", 100) + agg.Record("latency", 500) + agg.Record("latency", 300) + + max, _ := agg.GetMax("latency") + assert.Equal(t, 500.0, max) +} + +func TestGetMin(t *testing.T) { + agg := NewMetricsAggregator(1*time.Hour, 100) + + agg.Record("latency", 100) + agg.Record("latency", 500) + agg.Record("latency", 50) + + min, _ := agg.GetMin("latency") + assert.Equal(t, 50.0, min) +} + +func TestGetLatest(t *testing.T) { + agg := NewMetricsAggregator(1*time.Hour, 100) + + agg.Record("latency", 100) + agg.Record("latency", 200) + + latest, _ := agg.GetLatest("latency") + assert.Equal(t, 200.0, latest.Value) +} + +func TestMultipleMetrics(t *testing.T) { + agg := NewMetricsAggregator(1*time.Hour, 100) + + agg.Record("latency", 100) + agg.Record("errors", 5) + agg.Record("throughput", 1000) + + names := agg.GetMetricNames() + assert.Equal(t, 3, len(names)) +} + +func TestClear(t *testing.T) { + agg := NewMetricsAggregator(1*time.Hour, 100) + + agg.Record("latency", 100) + agg.Clear() + + names := agg.GetMetricNames() + assert.Equal(t, 0, len(names)) +} + +func TestGetCountInRange(t *testing.T) { + agg := NewMetricsAggregator(1*time.Hour, 100) + + now := time.Now() + agg.Record("latency", 100) + agg.Record("latency", 200) + + count, _ := agg.GetCountInRange("latency", now.Add(-1*time.Minute), now.Add(1*time.Minute)) + assert.Equal(t, 2, count) +} + +func TestNotFoundError(t *testing.T) { + agg := NewMetricsAggregator(1*time.Hour, 100) + + _, err := agg.GetPercentile("nonexistent", 50) + assert.Error(t, err) + + _, err = agg.GetAverage("nonexistent") + assert.Error(t, err) + + _, err = agg.GetLatest("nonexistent") + assert.Error(t, err) +} + +func TestMaxSize(t *testing.T) { + agg := NewMetricsAggregator(1*time.Hour, 5) + + for i := 0; i < 10; i++ { + agg.Record("latency", float64(i)) + } + + series := agg.GetTimeSeries("latency") + assert.Equal(t, 5, len(series)) +} + +func BenchmarkRecord(b *testing.B) { + agg := NewMetricsAggregator(1*time.Hour, 1000) + + b.ResetTimer() + for i := 0; i < b.N; i++ { + agg.Record("latency", float64(i)) + } +} diff --git a/internal/search/workflow_search.go b/internal/search/workflow_search.go new file mode 100644 index 0000000..bf903d3 --- /dev/null +++ b/internal/search/workflow_search.go @@ -0,0 +1,234 @@ +package search + +import ( + "fmt" + "regexp" + "strings" + "sync" + "time" +) + +// WorkflowEntry represents an indexed workflow +type WorkflowEntry struct { + ID string + Name string + Status string + CreatedAt time.Time + UpdatedAt time.Time + Tags []string + Content string + Assignee string +} + +// WorkflowSearch provides full-text search and filtering +type WorkflowSearch struct { + mu sync.RWMutex + entries map[string]*WorkflowEntry + index map[string][]string // word -> workflow IDs + filters map[string]interface{} +} + +// NewWorkflowSearch creates a new workflow search index +func NewWorkflowSearch() *WorkflowSearch { + return &WorkflowSearch{ + entries: make(map[string]*WorkflowEntry), + index: make(map[string][]string), + filters: make(map[string]interface{}), + } +} + +// Index adds a workflow to the search index +func (ws *WorkflowSearch) Index(entry *WorkflowEntry) error { + if entry.ID == "" { + return fmt.Errorf("workflow ID required") + } + + ws.mu.Lock() + defer ws.mu.Unlock() + + ws.entries[entry.ID] = entry + + // Index content + words := strings.Fields(strings.ToLower(entry.Content + " " + entry.Name)) + for _, word := range words { + // Remove punctuation + clean := strings.Trim(word, ".,!?;:") + if clean != "" { + ws.index[clean] = append(ws.index[clean], entry.ID) + } + } + + return nil +} + +// Search performs full-text search +func (ws *WorkflowSearch) Search(query string) []*WorkflowEntry { + ws.mu.RLock() + defer ws.mu.RUnlock() + + query = strings.ToLower(query) + matches := make(map[string]int) + + words := strings.Fields(query) + for _, word := range words { + if ids, exists := ws.index[word]; exists { + for _, id := range ids { + matches[id]++ + } + } + } + + // Sort by match count + result := make([]*WorkflowEntry, 0) + for id := range matches { + if entry, exists := ws.entries[id]; exists { + result = append(result, entry) + } + } + + return result +} + +// FilterByStatus filters workflows by status +func (ws *WorkflowSearch) FilterByStatus(status string) []*WorkflowEntry { + ws.mu.RLock() + defer ws.mu.RUnlock() + + result := make([]*WorkflowEntry, 0) + for _, entry := range ws.entries { + if entry.Status == status { + result = append(result, entry) + } + } + + return result +} + +// FilterByAssignee filters workflows by assignee +func (ws *WorkflowSearch) FilterByAssignee(assignee string) []*WorkflowEntry { + ws.mu.RLock() + defer ws.mu.RUnlock() + + result := make([]*WorkflowEntry, 0) + for _, entry := range ws.entries { + if entry.Assignee == assignee { + result = append(result, entry) + } + } + + return result +} + +// FilterByTag filters workflows by tag +func (ws *WorkflowSearch) FilterByTag(tag string) []*WorkflowEntry { + ws.mu.RLock() + defer ws.mu.RUnlock() + + result := make([]*WorkflowEntry, 0) + for _, entry := range ws.entries { + for _, t := range entry.Tags { + if t == tag { + result = append(result, entry) + break + } + } + } + + return result +} + +// FilterByDateRange filters workflows by date range +func (ws *WorkflowSearch) FilterByDateRange(start, end time.Time) []*WorkflowEntry { + ws.mu.RLock() + defer ws.mu.RUnlock() + + result := make([]*WorkflowEntry, 0) + for _, entry := range ws.entries { + if entry.CreatedAt.After(start) && entry.CreatedAt.Before(end) { + result = append(result, entry) + } + } + + return result +} + +// SearchRegex performs regex search on content +func (ws *WorkflowSearch) SearchRegex(pattern string) ([]*WorkflowEntry, error) { + re, err := regexp.Compile(pattern) + if err != nil { + return nil, err + } + + ws.mu.RLock() + defer ws.mu.RUnlock() + + result := make([]*WorkflowEntry, 0) + for _, entry := range ws.entries { + if re.MatchString(entry.Content) || re.MatchString(entry.Name) { + result = append(result, entry) + } + } + + return result, nil +} + +// SaveFilter saves a named filter +func (ws *WorkflowSearch) SaveFilter(name string, filter interface{}) { + ws.mu.Lock() + defer ws.mu.Unlock() + + ws.filters[name] = filter +} + +// GetFilter retrieves a saved filter +func (ws *WorkflowSearch) GetFilter(name string) (interface{}, bool) { + ws.mu.RLock() + defer ws.mu.RUnlock() + + filter, exists := ws.filters[name] + return filter, exists +} + +// GetAll returns all workflows +func (ws *WorkflowSearch) GetAll() []*WorkflowEntry { + ws.mu.RLock() + defer ws.mu.RUnlock() + + result := make([]*WorkflowEntry, 0, len(ws.entries)) + for _, entry := range ws.entries { + result = append(result, entry) + } + + return result +} + +// GetByID retrieves a workflow by ID +func (ws *WorkflowSearch) GetByID(id string) (*WorkflowEntry, bool) { + ws.mu.RLock() + defer ws.mu.RUnlock() + + entry, exists := ws.entries[id] + return entry, exists +} + +// Delete removes a workflow from the index +func (ws *WorkflowSearch) Delete(id string) error { + ws.mu.Lock() + defer ws.mu.Unlock() + + if _, exists := ws.entries[id]; !exists { + return fmt.Errorf("workflow not found: %s", id) + } + + delete(ws.entries, id) + return nil +} + +// Clear clears the entire index +func (ws *WorkflowSearch) Clear() { + ws.mu.Lock() + defer ws.mu.Unlock() + + ws.entries = make(map[string]*WorkflowEntry) + ws.index = make(map[string][]string) +} diff --git a/internal/search/workflow_search_test.go b/internal/search/workflow_search_test.go new file mode 100644 index 0000000..6cf08ea --- /dev/null +++ b/internal/search/workflow_search_test.go @@ -0,0 +1,196 @@ +package search + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +func TestIndex(t *testing.T) { + ws := NewWorkflowSearch() + + entry := &WorkflowEntry{ + ID: "wf-1", + Name: "Deploy Service", + Status: "completed", + Content: "deployment task", + } + + err := ws.Index(entry) + assert.NoError(t, err) + + retrieved, exists := ws.GetByID("wf-1") + assert.True(t, exists) + assert.Equal(t, "Deploy Service", retrieved.Name) +} + +func TestSearch(t *testing.T) { + ws := NewWorkflowSearch() + + ws.Index(&WorkflowEntry{ + ID: "wf-1", + Name: "Deploy Service", + Content: "deployment production", + }) + ws.Index(&WorkflowEntry{ + ID: "wf-2", + Name: "Build Docker", + Content: "docker image", + }) + + results := ws.Search("deployment") + assert.Equal(t, 1, len(results)) + assert.Equal(t, "wf-1", results[0].ID) +} + +func TestFilterByStatus(t *testing.T) { + ws := NewWorkflowSearch() + + ws.Index(&WorkflowEntry{ID: "wf-1", Status: "completed"}) + ws.Index(&WorkflowEntry{ID: "wf-2", Status: "running"}) + ws.Index(&WorkflowEntry{ID: "wf-3", Status: "completed"}) + + results := ws.FilterByStatus("completed") + assert.Equal(t, 2, len(results)) +} + +func TestFilterByAssignee(t *testing.T) { + ws := NewWorkflowSearch() + + ws.Index(&WorkflowEntry{ID: "wf-1", Assignee: "alice"}) + ws.Index(&WorkflowEntry{ID: "wf-2", Assignee: "bob"}) + ws.Index(&WorkflowEntry{ID: "wf-3", Assignee: "alice"}) + + results := ws.FilterByAssignee("alice") + assert.Equal(t, 2, len(results)) +} + +func TestFilterByTag(t *testing.T) { + ws := NewWorkflowSearch() + + ws.Index(&WorkflowEntry{ + ID: "wf-1", + Tags: []string{"production", "critical"}, + }) + ws.Index(&WorkflowEntry{ + ID: "wf-2", + Tags: []string{"staging"}, + }) + + results := ws.FilterByTag("production") + assert.Equal(t, 1, len(results)) +} + +func TestFilterByDateRange(t *testing.T) { + ws := NewWorkflowSearch() + + now := time.Now() + + ws.Index(&WorkflowEntry{ + ID: "wf-1", + CreatedAt: now.Add(-1 * time.Hour), + }) + ws.Index(&WorkflowEntry{ + ID: "wf-2", + CreatedAt: now.Add(-24 * time.Hour), + }) + + results := ws.FilterByDateRange(now.Add(-2*time.Hour), now) + assert.Equal(t, 1, len(results)) +} + +func TestSearchRegex(t *testing.T) { + ws := NewWorkflowSearch() + + ws.Index(&WorkflowEntry{ + ID: "wf-1", + Content: "error 404 not found", + }) + ws.Index(&WorkflowEntry{ + ID: "wf-2", + Content: "success 200 ok", + }) + + results, err := ws.SearchRegex("error.*404") + assert.NoError(t, err) + assert.Equal(t, 1, len(results)) +} + +func TestSaveAndGetFilter(t *testing.T) { + ws := NewWorkflowSearch() + + filter := map[string]interface{}{"status": "completed"} + ws.SaveFilter("completed-only", filter) + + retrieved, exists := ws.GetFilter("completed-only") + assert.True(t, exists) + assert.NotNil(t, retrieved) +} + +func TestGetAll(t *testing.T) { + ws := NewWorkflowSearch() + + ws.Index(&WorkflowEntry{ID: "wf-1"}) + ws.Index(&WorkflowEntry{ID: "wf-2"}) + ws.Index(&WorkflowEntry{ID: "wf-3"}) + + all := ws.GetAll() + assert.Equal(t, 3, len(all)) +} + +func TestDelete(t *testing.T) { + ws := NewWorkflowSearch() + + ws.Index(&WorkflowEntry{ID: "wf-1"}) + ws.Delete("wf-1") + + _, exists := ws.GetByID("wf-1") + assert.False(t, exists) +} + +func TestClear(t *testing.T) { + ws := NewWorkflowSearch() + + ws.Index(&WorkflowEntry{ID: "wf-1"}) + ws.Index(&WorkflowEntry{ID: "wf-2"}) + ws.Clear() + + all := ws.GetAll() + assert.Equal(t, 0, len(all)) +} + +func TestMultiwordSearch(t *testing.T) { + ws := NewWorkflowSearch() + + ws.Index(&WorkflowEntry{ + ID: "wf-1", + Content: "deploy service to production", + }) + + results := ws.Search("deploy service") + assert.Equal(t, 1, len(results)) +} + +func TestCaseSensitivity(t *testing.T) { + ws := NewWorkflowSearch() + + ws.Index(&WorkflowEntry{ + ID: "wf-1", + Content: "Deploy Service Production", + }) + + results := ws.Search("deploy") + assert.Equal(t, 1, len(results)) +} + +func TestIndexError(t *testing.T) { + ws := NewWorkflowSearch() + + entry := &WorkflowEntry{ + Name: "No ID", + } + + err := ws.Index(entry) + assert.Error(t, err) +} diff --git a/internal/visualization/dag_renderer.go b/internal/visualization/dag_renderer.go new file mode 100644 index 0000000..190f02e --- /dev/null +++ b/internal/visualization/dag_renderer.go @@ -0,0 +1,237 @@ +package visualization + +import ( + "fmt" + "strings" +) + +// TaskNode represents a task in the DAG +type TaskNode struct { + ID string + Status string // pending, running, completed, failed + Duration float64 + Critical bool +} + +// DAGRenderer renders workflow dependency graphs +type DAGRenderer struct { + nodes map[string]*TaskNode + edges map[string][]string +} + +// NewDAGRenderer creates a new DAG renderer +func NewDAGRenderer() *DAGRenderer { + return &DAGRenderer{ + nodes: make(map[string]*TaskNode), + edges: make(map[string][]string), + } +} + +// AddNode adds a task node +func (dr *DAGRenderer) AddNode(id, status string, duration float64) { + dr.nodes[id] = &TaskNode{ + ID: id, + Status: status, + Duration: duration, + } +} + +// AddEdge adds a dependency edge +func (dr *DAGRenderer) AddEdge(from, to string) error { + if _, exists := dr.nodes[from]; !exists { + return fmt.Errorf("source node not found: %s", from) + } + if _, exists := dr.nodes[to]; !exists { + return fmt.Errorf("target node not found: %s", to) + } + + dr.edges[from] = append(dr.edges[from], to) + return nil +} + +// MarkCriticalPath marks nodes on the critical path +func (dr *DAGRenderer) MarkCriticalPath(nodes []string) error { + for _, nodeID := range nodes { + if node, exists := dr.nodes[nodeID]; exists { + node.Critical = true + } else { + return fmt.Errorf("node not found: %s", nodeID) + } + } + return nil +} + +// RenderDOT generates DOT format for Graphviz +func (dr *DAGRenderer) RenderDOT() string { + var buf strings.Builder + + buf.WriteString("digraph WorkflowDAG {\n") + buf.WriteString(" rankdir=LR;\n") + buf.WriteString(" node [shape=box];\n\n") + + // Render nodes + for _, node := range dr.nodes { + color := "lightgray" + if node.Critical { + color = "red" + } else if node.Status == "completed" { + color = "lightgreen" + } else if node.Status == "failed" { + color = "lightcoral" + } else if node.Status == "running" { + color = "lightyellow" + } + + label := fmt.Sprintf("%s\\n%.0fms", node.ID, node.Duration) + buf.WriteString(fmt.Sprintf(" \"%s\" [label=\"%s\", fillcolor=%s, style=filled];\n", + node.ID, label, color)) + } + + buf.WriteString("\n") + + // Render edges + for from, tos := range dr.edges { + for _, to := range tos { + buf.WriteString(fmt.Sprintf(" \"%s\" -> \"%s\";\n", from, to)) + } + } + + buf.WriteString("}\n") + return buf.String() +} + +// RenderHTML generates a simple HTML visualization +func (dr *DAGRenderer) RenderHTML() string { + var buf strings.Builder + + buf.WriteString("
\n") + buf.WriteString("| Task ID | Status | Duration (ms) | Critical Path |
|---|---|---|---|
| %s | %s | %.0f | %s |