orchestra/internal/store/store.go

package store

import (
	"bufio"
	"encoding/json"
	"errors"
	"fmt"
	"orchestra/internal/authz"
	"orchestra/internal/domain"
	"os"
	"path/filepath"
	"sync"
	"time"
)

type Store struct {
	mu       sync.Mutex
	path     string
	cas      string
	events   []domain.Event
	tasks    map[string]domain.Task
	external map[string]string
	snapshot string
	seq      uint64
}

func Open(dir string) (*Store, error) {
	if err := os.MkdirAll(dir, 0755); err != nil {
		return nil, err
	}
	s := &Store{path: filepath.Join(dir, "events.jsonl"), cas: filepath.Join(dir, "cas"), snapshot: filepath.Join(dir, "snapshot.json"), tasks: map[string]domain.Task{}, external: map[string]string{}}
	if err := os.MkdirAll(s.cas, 0755); err != nil {
		return nil, err
	}
	var snapshotSeq uint64
	if b, readErr := os.ReadFile(s.snapshot); readErr == nil {
		var snap struct {
			Seq   uint64        `json:"seq"`
			Tasks []domain.Task `json:"tasks"`
		}
		if json.Unmarshal(b, &snap) != nil {
			return nil, fmt.Errorf("invalid snapshot")
		}
		for _, t := range snap.Tasks {
			s.tasks[t.ID] = t
			s.external[t.Source+"\x00"+t.ExternalID] = t.ID
		}
		snapshotSeq = snap.Seq
		// Continue event numbering after the snapshot. Without restoring this
		// cursor, the first append after a restart reused sequence 1 and made
		// the append-only log unreplayable.
		s.seq = snapshotSeq
	} else if !errors.Is(readErr, os.ErrNotExist) {
		return nil, readErr
	}
	f, err := os.Open(s.path)
	if os.IsNotExist(err) {
		return s, nil
	}
	if err != nil {
		return nil, err
	}
	defer f.Close()
	sc := bufio.NewScanner(f)
	var expected uint64 = snapshotSeq + 1
	for sc.Scan() {
		var e domain.Event
		if err := json.Unmarshal(sc.Bytes(), &e); err == nil {
			if err := domain.ValidateEvent(e); err != nil {
				return nil, err
			}
			if e.Seq < expected {
				continue
			}
			if e.Seq != expected {
				return nil, fmt.Errorf("event sequence gap: got %d, want %d", e.Seq, expected)
			}
			if t, ok := s.tasks[e.TaskID]; ok && e.Version != t.Version+1 {
				return nil, domain.ErrConflict
			}
			if _, ok := s.tasks[e.TaskID]; !ok && e.Type != "TaskCreated" {
				return nil, domain.ErrNotFound
			}
			s.events = append(s.events, e)
			s.seq = e.Seq
			if err := s.apply(e); err != nil {
				return nil, err
			}
			expected++
		} else {
			return nil, err
		}
	}
	return s, sc.Err()
}
func (s *Store) apply(e domain.Event) error {
	var p map[string]any
	if err := json.Unmarshal(e.Payload, &p); err != nil {
		return err
	}
	t := s.tasks[e.TaskID]
	if e.Type == "QuotaReported" || e.Type == "StandupAdvisory" || e.Type == "ApprovalGranted" || e.Type == "ApprovalDenied" {
		return nil
	}
	switch e.Type {
	case "TaskCreated":
		if err := domain.ValidateCreated(p); err != nil {
			return err
		}
		t = domain.Task{ID: e.TaskID, Source: p["source"].(string), ExternalID: p["external_id"].(string), Project: p["project"].(string), State: domain.StateQueued}
		if v, ok := p["parent"].(string); ok {
			t.Parent = v
		}
		if v, ok := p["inherent_priority"].(float64); ok {
			t.InherentPriority = int(v)
		}
		if v, ok := p["due"].(string); ok {
			if d, err := time.Parse(time.RFC3339, v); err == nil {
				t.Due = &d
			}
		}
		if v, ok := p["estimate"].(map[string]any); ok {
			t.Estimate = &domain.Estimate{}
			t.Estimate.Value, _ = v["value"].(float64)
			t.Estimate.Who, _ = v["who"].(string)
			t.Estimate.Confidence, _ = v["confidence"].(float64)
		}
		if v, ok := p["capability"].([]any); ok {
			for _, x := range v {
				if z, ok := x.(string); ok {
					t.Capability = append(t.Capability, z)
				}
			}
		}
		if v, ok := p["title"].(string); ok {
			t.Title = v
		}
		if v, ok := p["description"].(string); ok {
			t.Description = v
		}
		s.external[t.Source+"\x00"+t.ExternalID] = t.ID
	case "TaskLeased":
		t.State = domain.StateLeased
		t.Lease = &domain.Lease{HarnessID: p["harness_id"].(string), Until: time.Unix(0, int64(p["until_ns"].(float64)))}
	case "TaskReleased":
		t.State = domain.StateQueued
		t.Lease = nil
		t.HandoffRef, _ = p["handoff_ref"].(string)
	case "TaskCompleted":
		t.State = domain.StateCompleted
		t.Lease = nil
	case "TaskFailed":
		t.State = domain.StateFailed
		t.Lease = nil
	case "TaskBlocked":
		t.State = domain.StateBlocked
		t.Lease = nil
		t.Blocker, _ = p["blocker"].(string)
		t.BlockedAt = e.At
		t.LastPaneID, _ = p["pane_id"].(string)
		t.LastHarness, _ = p["harness_id"].(string)
		t.PaneState, _ = p["pane_state"].(string)
		if t.PaneState == "" {
			t.PaneState = "unknown"
		}
	case "TaskAmended":
		if v, ok := p["title"].(string); ok {
			t.Title = v
		}
		if v, ok := p["description"].(string); ok {
			t.Description = v
		}
		if v, ok := p["inherent_priority"].(float64); ok {
			t.InherentPriority = int(v)
		}
		if v, ok := p["due"].(string); ok {
			if d, err := time.Parse(time.RFC3339, v); err == nil {
				t.Due = &d
			}
		}
	case "TaskCorrected":
		if v, ok := p["title"].(string); ok {
			t.Title = v
		}
		if v, ok := p["description"].(string); ok {
			t.Description = v
		}
		if v, ok := p["inherent_priority"].(float64); ok {
			t.InherentPriority = int(v)
		}
		if v, ok := p["due"].(string); ok {
			if d, err := time.Parse(time.RFC3339, v); err == nil {
				t.Due = &d
			}
		}
		if v, ok := p["state"].(string); ok {
			t.State = domain.TaskState(v)
			if t.State != domain.StateLeased {
				t.Lease = nil
			}
		}
	}
	t.Version = e.Version
	s.tasks[e.TaskID] = t
	return nil
}
func (s *Store) Append(e domain.Event) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if e.At.IsZero() {
		e.At = time.Now().UTC()
	}
	if e.Seq == 0 {
		e.Seq = s.seq + 1
	}
	if e.SchemaVersion == 0 {
		e.SchemaVersion = domain.CurrentEventSchema
	}
	if err := domain.ValidateEvent(e); err != nil {
		return err
	}
	// Enforced once, at the append boundary, per spec §7.1/invariant 4 — every
	// producer (HTTP handler, router, coordinator, provider, federation relay)
	// must declare its Surface here; there is no separate in-process bypass.
	if err := authz.AuthorizeEvent(authz.Surface(e.Surface), e.Type); err != nil {
		return err
	}
	if e.Type == "TaskCreated" {
		var p map[string]any
		if err := json.Unmarshal(e.Payload, &p); err != nil {
			return err
		}
		if id := s.external[p["source"].(string)+"\x00"+p["external_id"].(string)]; id != "" {
			return domain.ErrDuplicate
		}
	}
	t, taskExists := s.tasks[e.TaskID]
	if taskExists && e.Version != t.Version+1 {
		return domain.ErrConflict
	}
	// Every optimistic lifecycle writer may carry its observed version. Enforce
	// it at the append boundary so non-HTTP producers receive the same CAS.
	var contract map[string]any
	if err := json.Unmarshal(e.Payload, &contract); err != nil {
		return err
	}
	if expected, ok := contract["expected_version"].(float64); ok {
		if expected != float64(int(expected)) || !taskExists || int(expected) != t.Version {
			return domain.ErrConflict
		}
	}
	if e.Type == "TaskLeased" {
		var p struct {
			ExpectedVersion *int `json:"expected_version"`
		}
		if err := json.Unmarshal(e.Payload, &p); err != nil {
			return err
		}
		if p.ExpectedVersion != nil && (t.Version != *p.ExpectedVersion) {
			return domain.ErrConflict
		}
	}
	if e.Type == "TaskCorrected" {
		var p map[string]any
		_ = json.Unmarshal(e.Payload, &p)
		corrects, _ := p["corrects"].(string)
		found := false
		for _, prior := range s.events {
			if prior.ID == corrects && prior.TaskID == e.TaskID {
				found = true
				break
			}
		}
		if !found {
			return fmt.Errorf("%w: corrects references unknown event %q for this task", domain.ErrInvalid, corrects)
		}
	}
	global := e.Type == "QuotaReported" || e.Type == "StandupAdvisory" || e.Type == "ApprovalGranted" || e.Type == "ApprovalDenied"
	if !taskExists && e.Type != "TaskCreated" && !global {
		return domain.ErrNotFound
	}
	if e.Type != "TaskCreated" && (e.Type == "TaskCompleted" || e.Type == "TaskBlocked" || e.Type == "TaskReleased") {
		var p map[string]any
		_ = json.Unmarshal(e.Payload, &p)
		for _, k := range []string{"handoff_ref", "report_ref"} {
			if ref, ok := p[k].(string); ok {
				if _, err := s.Artifact(ref); err != nil {
					return fmt.Errorf("%w: missing artifact %s", domain.ErrInvalid, ref)
				}
			}
		}
	}
	if err := s.apply(e); err != nil {
		return err
	}
	f, err := os.OpenFile(s.path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0644)
	if err != nil {
		return err
	}
	defer f.Close()
	b, _ := json.Marshal(e)
	if _, err = f.Write(append(b, '\n')); err != nil {
		return err
	}
	if err = f.Sync(); err != nil {
		return err
	}
	s.events = append(s.events, e)
	s.seq = e.Seq
	if err := s.writeSnapshot(); err != nil {
		return err
	}
	return nil
}
func (s *Store) writeSnapshot() error {
	tasks := make([]domain.Task, 0, len(s.tasks))
	for _, t := range s.tasks {
		tasks = append(tasks, t)
	}
	b, err := json.Marshal(struct {
		Seq   uint64        `json:"seq"`
		Tasks []domain.Task `json:"tasks"`
	}{s.seq, tasks})
	if err != nil {
		return err
	}
	tmp := s.snapshot + ".tmp"
	if err = os.WriteFile(tmp, b, 0644); err != nil {
		return err
	}
	return os.Rename(tmp, s.snapshot)
}
func (s *Store) Tasks() []domain.Task {
	s.mu.Lock()
	defer s.mu.Unlock()
	out := make([]domain.Task, 0, len(s.tasks))
	for _, t := range s.tasks {
		out = append(out, t)
	}
	return out
}
func (s *Store) Events(since uint64) []domain.Event {
	s.mu.Lock()
	defer s.mu.Unlock()
	var out []domain.Event
	for _, e := range s.events {
		if e.Seq > since {
			out = append(out, e)
		}
	}
	return out
}
func (s *Store) PutArtifact(b []byte) (string, error) {
	h := domain.Hash(b)
	p := filepath.Join(s.cas, h)
	if _, err := os.Stat(p); errors.Is(err, os.ErrNotExist) {
		if err = os.WriteFile(p, b, 0644); err != nil {
			return "", err
		}
	}
	return h, nil
}

// Artifact returns a CAS artifact after verifying its content address.
func (s *Store) Artifact(ref string) ([]byte, error) {
	if len(ref) != 64 {
		return nil, fmt.Errorf("%w: invalid artifact reference", domain.ErrInvalid)
	}
	b, err := os.ReadFile(filepath.Join(s.cas, ref))
	if err != nil {
		return nil, err
	}
	if domain.Hash(b) != ref {
		return nil, fmt.Errorf("%w: corrupt artifact", domain.ErrInvalid)
	}
	return b, nil
}

func (s *Store) Task(id string) (domain.Task, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	t, ok := s.tasks[id]
	return t, ok
}

// TaskBySource resolves the task ingested for a given (source, external_id)
// pair — the dedup key Append.ErrDuplicate rejects re-ingestion against.
func (s *Store) TaskBySource(source, externalID string) (domain.Task, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	id := s.external[source+"\x00"+externalID]
	if id == "" {
		return domain.Task{}, false
	}
	t, ok := s.tasks[id]
	return t, ok
}

func (s *Store) Lease(id, harness string, ttl time.Duration) (domain.Event, error) {
	if ttl <= 0 {
		return domain.Event{}, fmt.Errorf("%w: ttl must be positive", domain.ErrInvalid)
	}
	t, ok := s.Task(id)
	if !ok {
		return domain.Event{}, domain.ErrNotFound
	}
	if t.State != domain.StateQueued {
		return domain.Event{}, domain.ErrConflict
	}
	payload := map[string]any{"harness_id": harness, "ttl": ttl.Seconds(), "until_ns": time.Now().Add(ttl).UnixNano(), "expected_version": t.Version}
	if t.HandoffRef != "" {
		payload["handoff_ref"] = t.HandoffRef
	}
	p, _ := json.Marshal(payload)
	e := domain.Event{ID: domain.NewID(), Type: "TaskLeased", TaskID: id, Version: t.Version + 1, Payload: p, Surface: string(authz.System)}
	return e, s.Append(e)
}

func (s *Store) ExpireLeases(now time.Time) ([]domain.Event, error) {
	var out []domain.Event
	for _, t := range s.Tasks() {
		if t.State == domain.StateLeased && t.Lease != nil && !t.Lease.Until.After(now) {
			p, _ := json.Marshal(map[string]any{"reason": "lease_expired", "harness_id": t.Lease.HarnessID})
			e := domain.Event{ID: domain.NewID(), Type: "TaskReleased", TaskID: t.ID, Version: t.Version + 1, Payload: p, Surface: string(authz.System)}
			if err := s.Append(e); err != nil {
				return out, err
			}
			out = append(out, e)
		}
	}
	return out, nil
}