Implement projections and operations

This commit is contained in:
kami
2026-07-26 19:14:46 +04:00
parent 9937cd5cd0
commit 8822e028bb
6 changed files with 148 additions and 7 deletions
+1 -1
View File
@@ -79,7 +79,7 @@ func ValidateEvent(e Event) error {
if e.Type == "" || e.TaskID == "" || len(e.Payload) == 0 || len(e.Payload) > 64*1024 {
return ErrInvalid
}
allowed := map[string]bool{"TaskCreated": true, "TaskLeased": true, "TaskReleased": true, "TaskCompleted": true, "TaskFailed": true, "TaskBlocked": true, "ApprovalRequested": true, "ApprovalGranted": true, "ApprovalDenied": true, "TaskAmended": true}
allowed := map[string]bool{"TaskCreated": true, "TaskLeased": true, "TaskReleased": true, "TaskCompleted": true, "TaskFailed": true, "TaskBlocked": true, "ApprovalRequested": true, "ApprovalGranted": true, "ApprovalDenied": true, "TaskAmended": true, "QuotaReported": true, "StandupAdvisory": true}
if !allowed[e.Type] {
return fmt.Errorf("%w: unknown type %q", ErrInvalid, e.Type)
}
+78
View File
@@ -0,0 +1,78 @@
// Package operations contains read-side projections and operational observability.
package operations
import (
"encoding/json"
"fmt"
"orchestra/internal/domain"
"os/exec"
"strings"
"time"
)
type Brief struct {
From, To time.Time `json:"from"`
Completed int `json:"completed"`
Failed int `json:"failed"`
Blocked int `json:"blocked"`
NeedsAttention []domain.Event `json:"needs_attention"`
Quota map[string]float64 `json:"quota_consumed"`
Git GitSync `json:"git_sync"`
}
type GitSync struct {
Branch, Head, Status string `json:"branch" json:"head" json:"status"`
}
// BuildBrief folds only events in [from,to]. It is deliberately read-only.
func BuildBrief(events []domain.Event, from, to time.Time, git GitSync) Brief {
b := Brief{From: from, To: to, Quota: map[string]float64{}, Git: git}
for _, e := range events {
if e.At.Before(from) || e.At.After(to) {
continue
}
var p map[string]any
_ = json.Unmarshal(e.Payload, &p)
switch e.Type {
case "TaskCompleted":
b.Completed++
case "TaskFailed":
b.Failed++
b.NeedsAttention = append(b.NeedsAttention, e)
case "TaskBlocked":
b.Blocked++
b.NeedsAttention = append(b.NeedsAttention, e)
case "ApprovalRequested":
b.NeedsAttention = append(b.NeedsAttention, e)
case "QuotaReported":
if h, ok := p["harness_id"].(string); ok {
if n, ok := p["consumed"].(float64); ok {
b.Quota[h] += n
}
}
}
}
return b
}
// GitState reports local checkout state for the brief. Git failures are visible,
// never silently interpreted as synchronized.
func GitState(dir string) GitSync {
g := GitSync{Status: "unavailable"}
run := func(args ...string) string {
out, err := exec.Command("git", append([]string{"-C", dir}, args...)...).Output()
if err != nil {
return ""
}
return strings.TrimSpace(string(out))
}
g.Branch, g.Head = run("branch", "--show-current"), run("rev-parse", "HEAD")
if s := run("status", "--porcelain"); s != "" {
g.Status = "dirty"
} else if g.Head != "" {
g.Status = "clean"
}
if g.Head == "" {
g.Status = fmt.Sprintf("git unavailable (%s)", dir)
}
return g
}
+18
View File
@@ -0,0 +1,18 @@
package operations
import (
"encoding/json"
"orchestra/internal/domain"
"testing"
"time"
)
func TestBuildBrief(t *testing.T) {
now := time.Now()
p, _ := json.Marshal(map[string]any{"harness_id": "cc", "consumed": 12})
es := []domain.Event{{Type: "TaskCompleted", At: now}, {Type: "TaskFailed", At: now}, {Type: "ApprovalRequested", At: now}, {Type: "QuotaReported", At: now, Payload: p}}
b := BuildBrief(es, now.Add(-time.Minute), now.Add(time.Minute), GitSync{Status: "clean"})
if b.Completed != 1 || b.Failed != 1 || len(b.NeedsAttention) != 2 || b.Quota["cc"] != 12 {
t.Fatalf("unexpected brief: %+v", b)
}
}