Implement provider layer adapters

2026-07-26 18:59:13 +04:00
parent 24ee81d538
commit 0a21e1bc2b
2 changed files with 239 additions and 15 deletions
@@ -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)
+}
@@ -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.