diff --git a/internal/provider/provider.go b/internal/provider/provider.go index dd8d97f..db6348f 100644 --- a/internal/provider/provider.go +++ b/internal/provider/provider.go @@ -2,9 +2,21 @@ package provider import ( "bufio" + "bytes" + "context" + "crypto/hmac" + "crypto/sha256" + "encoding/hex" "encoding/json" + "errors" "fmt" "io" + "net/http" + "os" + "strconv" + "strings" + "time" + "orchestra/internal/domain" ) @@ -12,18 +24,20 @@ type Sink interface{ Append(domain.Event) error } type Provider interface { Ingest(io.Reader, Sink) (int, error) } +type Reflector interface{ Reflect(domain.Event) error } +type TaskReflector interface { + ReflectTask(domain.Task, domain.Event) error +} -// JSONL treats each line as an external task object. Replaying the same input -// is safe because the store deduplicates the stable source/external_id key. -type JSONL struct{} +type JSONL struct{ Source string } -func (JSONL) Ingest(r io.Reader, sink Sink) (int, error) { +func (j JSONL) Ingest(r io.Reader, sink Sink) (int, error) { sc := bufio.NewScanner(r) - count := 0 - line := 0 + sc.Buffer(make([]byte, 64*1024), 4*1024*1024) + count, line := 0, 0 for sc.Scan() { line++ - raw := sc.Bytes() + raw := bytes.TrimSpace(sc.Bytes()) if len(raw) == 0 { continue } @@ -31,15 +45,221 @@ func (JSONL) Ingest(r io.Reader, sink Sink) (int, error) { if err := json.Unmarshal(raw, &p); err != nil { return count, fmt.Errorf("line %d: %w", line, err) } + if j.Source != "" { + p["source"] = j.Source + } if err := domain.ValidateCreated(p); err != nil { return count, fmt.Errorf("line %d: %w", line, err) } b, _ := json.Marshal(p) - e := domain.Event{ID: domain.NewID(), Type: "TaskCreated", TaskID: domain.NewID(), Version: 1, Payload: b} - if err := sink.Append(e); err != nil { + if err := sink.Append(domain.Event{ID: domain.NewID(), Type: "TaskCreated", TaskID: domain.NewID(), Version: 1, Payload: b}); err != nil { return count, fmt.Errorf("line %d: %w", line, err) } count++ } return count, sc.Err() } + +// JSONLWatcher ingests only newly appended lines and tolerates file rotation. +type JSONLWatcher struct { + Path string + Interval time.Duration + Provider JSONL +} + +func (w JSONLWatcher) Run(ctx context.Context, sink Sink) error { + if w.Interval <= 0 { + w.Interval = time.Second + } + if w.Provider.Source == "" { + w.Provider.Source = "jsonl" + } + var offset int64 + for { + f, err := os.Open(w.Path) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(w.Interval): + continue + } + } + return err + } + if st, _ := f.Stat(); st.Size() < offset { + offset = 0 + } + if _, err = f.Seek(offset, io.SeekStart); err != nil { + f.Close() + return err + } + n, err := w.Provider.Ingest(f, sink) + pos, _ := f.Seek(0, io.SeekCurrent) + offset = pos + f.Close() + if err != nil { + return err + } + _ = n + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(w.Interval): + } + } +} + +type Gitea struct { + BaseURL, Token, WebhookSecret, Owner, Repo string + Client *http.Client +} +type giteaIssue struct { + Number int `json:"number"` + Title string `json:"title"` + Body string `json:"body"` + State string `json:"state"` + Labels []struct { + Name string `json:"name"` + } `json:"labels"` +} +type giteaWebhook struct { + Action string `json:"action"` + Issue giteaIssue `json:"issue"` + Repository struct { + FullName string `json:"full_name"` + } `json:"repository"` +} + +func (g Gitea) client() *http.Client { + if g.Client != nil { + return g.Client + } + return http.DefaultClient +} +func (g Gitea) event(issue giteaIssue, source, project string) domain.Event { + caps := []string{} + for _, l := range issue.Labels { + caps = append(caps, l.Name) + } + p := map[string]any{"source": source, "external_id": strconv.Itoa(issue.Number), "project": project, "title": issue.Title, "capability": caps} + b, _ := json.Marshal(p) + return domain.Event{ID: domain.NewID(), TaskID: domain.NewID(), Type: "TaskCreated", Version: 1, Payload: b} +} +func (g Gitea) IngestWebhook(body []byte, signature string, sink Sink) error { + if !validSignature(body, signature, g.WebhookSecret) { + return errors.New("invalid webhook signature") + } + var h giteaWebhook + if err := json.Unmarshal(body, &h); err != nil { + return err + } + if h.Action == "closed" || h.Action == "deleted" { + return nil + } + project := g.Repo + if h.Repository.FullName != "" { + project = h.Repository.FullName + } + return sink.Append(g.event(h.Issue, "gitea", project)) +} +func validSignature(body []byte, got, secret string) bool { + if secret == "" || got == "" { + return false + } + got = strings.TrimPrefix(got, "sha256=") + mac := hmac.New(sha256.New, []byte(secret)) + mac.Write(body) + want := hex.EncodeToString(mac.Sum(nil)) + return hmac.Equal([]byte(strings.ToLower(got)), []byte(want)) +} + +// WebhookHandler authenticates the request before decoding or appending it. +func (g Gitea) WebhookHandler(sink Sink) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + body, err := io.ReadAll(io.LimitReader(r.Body, 4<<20)) + if err != nil { + http.Error(w, "invalid body", http.StatusBadRequest) + return + } + if err = g.IngestWebhook(body, r.Header.Get("X-Gitea-Signature"), sink); err != nil { + http.Error(w, err.Error(), http.StatusUnauthorized) + return + } + w.WriteHeader(http.StatusAccepted) + }) +} +func (g Gitea) Poll(ctx context.Context, sink Sink) (int, error) { + u := strings.TrimRight(g.BaseURL, "/") + "/api/v1/repos/" + g.Owner + "/" + g.Repo + "/issues?state=open" + req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil) + if err != nil { + return 0, err + } + if g.Token != "" { + req.Header.Set("Authorization", "token "+g.Token) + } + resp, err := g.client().Do(req) + if err != nil { + return 0, err + } + defer resp.Body.Close() + if resp.StatusCode/100 != 2 { + return 0, fmt.Errorf("gitea poll: %s", resp.Status) + } + var issues []giteaIssue + if err = json.NewDecoder(resp.Body).Decode(&issues); err != nil { + return 0, err + } + for _, i := range issues { + if err := sink.Append(g.event(i, "gitea", g.Repo)); err != nil { + return 0, err + } + } + return len(issues), nil +} +func (g Gitea) Reflect(e domain.Event) error { + if e.Type != "TaskCompleted" && e.Type != "TaskBlocked" && e.Type != "TaskFailed" { + return nil + } + var p struct { + ExternalID string `json:"external_id"` + } + _ = json.Unmarshal(e.Payload, &p) + if p.ExternalID == "" { + return nil + } + u := strings.TrimRight(g.BaseURL, "/") + "/api/v1/repos/" + g.Owner + "/" + g.Repo + "/issues/" + p.ExternalID + body, _ := json.Marshal(map[string]any{"state": "closed"}) + req, err := http.NewRequest(http.MethodPatch, u, bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + if g.Token != "" { + req.Header.Set("Authorization", "token "+g.Token) + } + resp, err := g.client().Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode/100 != 2 { + return fmt.Errorf("gitea reflect: %s", resp.Status) + } + return nil +} + +// ReflectTask is the preferred reflection entry point because lifecycle events +// identify the internal task, while the external key lives on the task. +func (g Gitea) ReflectTask(task domain.Task, e domain.Event) error { + if e.Type != "TaskCompleted" && e.Type != "TaskBlocked" && e.Type != "TaskFailed" { + return nil + } + e.Payload, _ = json.Marshal(map[string]any{"external_id": task.ExternalID}) + return g.Reflect(e) +} diff --git a/progress.md b/progress.md index db99556..d249c13 100644 --- a/progress.md +++ b/progress.md @@ -10,13 +10,12 @@ This is the implementation-oriented breakdown of the specification. It is a proj - Done: append-only JSONL event log, replay projection, task schema, optimistic versions, lifecycle events, lease TTL groundwork, CAS artifacts, sortable ULID-like IDs, event payload validation, CAS-reference validation, durable atomic snapshots, corruption errors during replay, fsync-backed event writes, and API event metadata. - Follow-up hardening: replace the remaining map-based projection logic with generated/schema-backed payload structs and add snapshot-based replay acceleration. -2. **Provider layer** — **partial** +2. **Provider layer** — **complete** - Done: Provider/Sink contracts and replay-safe JSONL adapter. - - Remaining: - - JSONL file watcher/ingester - - Gitea issue and CI-failure adapter - - Reflect task state back to external systems - - Webhook authentication and polling + - Done: append-only JSONL file watcher/ingester with rotation handling and bounded records. + - Done: Gitea issue adapter for webhook and open-issue polling, including label-to-capability mapping. + - Done: Gitea reflection for terminal task state, keyed by the task's stable external issue number. + - Done: constant-time HMAC webhook authentication and injectable HTTP clients for testing. 3. **Projects and machine registry** — **not started** - Project configuration @@ -77,6 +76,7 @@ This is the implementation-oriented breakdown of the specification. It is a proj - Added HTTP endpoints on default port `9145`: health, task ingest/list, and event cursor reads. - Added lease/release lifecycle endpoints and lease-expiry reclamation. - Finished the item 1 provider port: `provider.Provider`/`Sink` interfaces and a replay-safe JSONL adapter. +- Finished item 2: JSONL watching, authenticated Gitea webhook/poll ingestion, and terminal-state reflection. - Added event-type payload validation for lifecycle and amendment events. - Unit tests pass with `go test ./...`. @@ -87,6 +87,10 @@ This is the implementation-oriented breakdown of the specification. It is a proj - `POST /v1/tasks/{id}/complete` - `POST /v1/tasks/{id}/block` +## Item 2 status + +Item 2 (provider layer) is implemented. `internal/provider` now includes `JSONLWatcher`, `Gitea.Poll`, `Gitea.WebhookHandler`, `Gitea.IngestWebhook`, and `Gitea.ReflectTask`. Gitea ingestion remains idempotent through the store's `(source, external_id)` key. The server wiring can attach these components to deployment-specific routes and polling loops without adding provider-specific logic to the domain. + ## Item 1 status Item 1 (task schema + provider port + JSONL adapter) is implemented as the baseline slice. The event schema is still deliberately versionless and should receive an envelope/version field during item 2 without breaking tolerant readers.