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
	}
	// A snapshot is a disposable read cache, never recovery authority. Loading
	// it before the log let a partially-written snapshot become a different
	// history than events.jsonl after a crash. Rebuild every projection from
	// the append-only, fsynced log instead.
	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 = 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 {
				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
		}
		if v, ok := p["acceptance"].([]any); ok {
			for _, item := range v {
				if text, ok := item.(string); ok {
					t.Acceptance = append(t.Acceptance, text)
				}
			}
		}
		if v, ok := p["quality_gate"].(string); ok {
			t.QualityGate = v
		}
		s.external[t.Source+"\x00"+t.ExternalID] = t.ID
	case "TaskLeased":
		t.State = domain.StateLeased
		t.LifecyclePhase = "lease_issued"
		t.LastError = ""
		epoch, _ := p["lease_epoch"].(string)
		if epoch == "" {
			// A pre-fencing event cannot safely be renewed by an old worker.
			// Deriving a stable token from the durable event identity makes the
			// recovered lease observable but non-renewable until it expires.
			epoch = "legacy:" + e.ID
		}
		t.Lease = &domain.Lease{HarnessID: p["harness_id"].(string), Epoch: epoch, Until: time.Unix(0, int64(p["until_ns"].(float64)))}
	case "TaskLeaseRenewed":
		epoch, _ := p["lease_epoch"].(string)
		if epoch == "" && t.Lease != nil {
			epoch = t.Lease.Epoch
		}
		t.Lease = &domain.Lease{HarnessID: p["harness_id"].(string), Epoch: epoch, Until: time.Unix(0, int64(p["until_ns"].(float64)))}
	case "TaskLaunchAcknowledged":
		t.LifecyclePhase = "started"
	case "TaskReleased":
		t.State = domain.StateQueued
		t.LifecyclePhase = "reclaimed"
		t.Lease = nil
		t.HandoffRef, _ = p["handoff_ref"].(string)
		t.ReleaseTransaction, _ = p["transaction_id"].(string)
		t.ReleaseAnchor, _ = p["anchor_sha"].(string)
		t.PickupTransaction, t.PickupLeaseVersion = "", 0
		if t.HandoffRef == "" {
			// A handoff-less release is a reclaim. Persist the retry decision
			// here so expiry, pane exit, and a worker NACK all use the same
			// crash-safe transition instead of router-local counters.
			t.Attempt++
			t.FailureClass, _ = p["failure_class"].(string)
			if t.FailureClass == "" {
				t.FailureClass, _ = p["reason"].(string)
			}
			t.NextRetryAt = e.At.Add(retryBackoff(t.Attempt))
		} else {
			t.NextRetryAt = time.Time{}
			t.FailureClass = ""
		}
	case "TaskPickupValidated":
		t.PickupTransaction, _ = p["transaction_id"].(string)
		if v, ok := p["lease_version"].(float64); ok {
			t.PickupLeaseVersion = int(v)
		}
	case "TaskCompleted":
		t.State = domain.StateCompleted
		t.Lease = nil
	case "TaskFailed":
		t.State = domain.StateFailed
		t.Lease = nil
	case "TaskBlocked", "TaskNeedsAttention":
		if e.Type == "TaskBlocked" {
			t.State = domain.StateBlocked
			t.Lease = nil
		} else {
			// Recovery diagnostics must not revoke the fenced owner. A late
			// completion is still valid only from this exact lease epoch.
			t.State = domain.StateNeedsAttention
		}
		t.Blocker, _ = p["blocker"].(string)
		t.BlockReason = domain.InferBlockReason(t.Blocker)
		if v, ok := p["block_reason"].(string); ok && domain.BlockReason(v).Valid() {
			t.BlockReason = domain.BlockReason(v)
		}
		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
			}
		}
	}
	if phase, ok := p["lifecycle_phase"].(string); ok && phase != "" {
		t.LifecyclePhase = phase
	}
	if last, ok := p["last_error"].(string); ok {
		t.LastError = last
	}
	// Terminal and release events may carry an owner-produced snapshot from
	// immediately before a worker/coordinator drops its live session mapping.
	// Preserve it independently of the current task state so historical task
	// pages never have to imply a pane is live just because its ID is known.
	if raw, ok := p["session_evidence"].(map[string]any); ok {
		var evidence domain.SessionEvidence
		if b, err := json.Marshal(raw); err == nil && json.Unmarshal(b, &evidence) == nil {
			t.LastSession = evidence
			if evidence.PaneID != "" {
				t.LastPaneID = evidence.PaneID
			}
			if evidence.HarnessID != "" {
				t.LastHarness = evidence.HarnessID
			}
			if evidence.PaneState != "" {
				t.PaneState = evidence.PaneState
			}
		}
	}
	t.Version = e.Version
	s.tasks[e.TaskID] = t
	return nil
}

func retryBackoff(attempt int) time.Duration {
	if attempt < 1 {
		attempt = 1
	}
	backoff := time.Minute
	for i := 1; i < attempt && backoff < 30*time.Minute; i++ {
		backoff *= 2
	}
	if backoff > 30*time.Minute {
		return 30 * time.Minute
	}
	return backoff
}
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 err := s.validateTransition(e, t, taskExists, contract); err != nil {
		return err
	}
	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 == "TaskNeedsAttention" || 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)
				}
			}
		}
	}
	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
	}
	// The event is the commit record. Do not expose a projection that cannot
	// be recovered from it after a power loss.
	if err := s.apply(e); err != nil {
		return err
	}
	s.events = append(s.events, e)
	s.seq = e.Seq
	// Snapshot failure does not roll back a committed event. Open always
	// rebuilds from the log, so leaving a stale cache is safe.
	_ = s.writeSnapshot()
	return nil
}

// validateTransition keeps lifecycle authority at the durable append
// boundary. A task may be completed/failed while queued by an external
// provider, but once a lease exists its owner and fencing epoch are required
// for every lifecycle mutation.
func (s *Store) validateTransition(e domain.Event, t domain.Task, exists bool, p map[string]any) error {
	if !exists {
		if e.Type != "TaskCreated" && e.Type != "QuotaReported" && e.Type != "StandupAdvisory" && e.Type != "ApprovalGranted" && e.Type != "ApprovalDenied" {
			return domain.ErrNotFound
		}
		return nil
	}
	if e.Type == "TaskLeased" && t.State != domain.StateQueued {
		return domain.ErrConflict
	}
	// Needs-attention is specifically a recoverable leased state, never a
	// second spelling of a terminal operator block on an unowned task.
	if e.Type == "TaskNeedsAttention" && ((t.State != domain.StateLeased && t.State != domain.StateNeedsAttention) || t.Lease == nil) {
		return domain.ErrConflict
	}
	if (t.State != domain.StateLeased && t.State != domain.StateNeedsAttention) || t.Lease == nil {
		return nil
	}
	switch e.Type {
	case "TaskLeaseRenewed", "TaskLaunchAcknowledged", "TaskReleased", "TaskPickupValidated", "TaskCompleted", "TaskBlocked", "TaskNeedsAttention", "TaskFailed":
		owner, _ := p["harness_id"].(string)
		epoch, _ := p["lease_epoch"].(string)
		// Expiry is the one coordinator-owned relinquish path. It still binds
		// the exact epoch that was observed when the timer fired.
		if e.Type == "TaskReleased" {
			if reason, _ := p["reason"].(string); (reason == "lease_expired" || reason == "pane_exited") && owner == t.Lease.HarnessID && epoch == t.Lease.Epoch {
				return nil
			}
		}
		if owner != t.Lease.HarnessID || epoch == "" || epoch != t.Lease.Epoch {
			return domain.ErrConflict
		}
	case "TaskCorrected":
		// Corrections may repair metadata while a task is leased, but cannot
		// smuggle in a lifecycle transition around the current fenced owner.
		if _, changesState := p["state"]; changesState {
			owner, _ := p["harness_id"].(string)
			epoch, _ := p["lease_epoch"].(string)
			if owner != t.Lease.HarnessID || epoch == "" || epoch != t.Lease.Epoch {
				return domain.ErrConflict
			}
		}
	}
	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"
	f, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0644)
	if err != nil {
		return err
	}
	if _, err = f.Write(b); err == nil {
		err = f.Sync()
	}
	if closeErr := f.Close(); err == nil {
		err = closeErr
	}
	if err != nil {
		return err
	}
	if err = os.Rename(tmp, s.snapshot); err != nil {
		return err
	}
	dir, err := os.Open(filepath.Dir(s.snapshot))
	if err != nil {
		return err
	}
	defer dir.Close()
	return dir.Sync()
}
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) {
		tmp := p + ".tmp-" + domain.NewID()
		f, openErr := os.OpenFile(tmp, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0644)
		if openErr != nil {
			return "", openErr
		}
		if _, err = f.Write(b); err == nil {
			err = f.Sync()
		}
		if closeErr := f.Close(); err == nil {
			err = closeErr
		}
		if err != nil {
			_ = os.Remove(tmp)
			return "", err
		}
		if err = os.Rename(tmp, p); err != nil && !errors.Is(err, os.ErrExist) {
			_ = os.Remove(tmp)
			return "", err
		}
		if !errors.Is(err, os.ErrExist) {
			dir, openErr := os.Open(s.cas)
			if openErr != nil {
				return "", openErr
			}
			syncErr := dir.Sync()
			closeErr := dir.Close()
			if syncErr != nil {
				return "", syncErr
			}
			if closeErr != nil {
				return "", closeErr
			}
		}
	}
	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, "lease_epoch": domain.NewID(), "ttl": ttl.Seconds(), "until_ns": time.Now().Add(ttl).UnixNano(), "expected_version": t.Version}
	if t.HandoffRef != "" {
		payload["handoff_ref"] = t.HandoffRef
		payload["transaction_id"] = t.ReleaseTransaction
		payload["anchor_sha"] = t.ReleaseAnchor
	}
	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)
}

// RenewLease atomically extends the current owner's lease. The observed task
// version is part of the request so an old worker can never renew a lease
// after release/reassignment.
func (s *Store) RenewLease(id, harness, epoch string, expectedVersion int, 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.StateLeased && t.State != domain.StateNeedsAttention) || t.Lease == nil || t.Lease.HarnessID != harness || t.Lease.Epoch != epoch || t.Version != expectedVersion {
		return domain.Event{}, domain.ErrConflict
	}
	p, _ := json.Marshal(map[string]any{"harness_id": harness, "lease_epoch": epoch, "until_ns": time.Now().Add(ttl).UnixNano(), "expected_version": expectedVersion})
	e := domain.Event{ID: domain.NewID(), Type: "TaskLeaseRenewed", 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 e, err := s.ExpireLease(t.ID, now); err != nil {
			if !errors.Is(err, domain.ErrConflict) {
				return out, err
			}
		} else if e.ID != "" {
			out = append(out, e)
		}
	}
	return out, nil
}

// ExpireLease releases exactly the observed lease if, and only if, its TTL
// has elapsed. Coordinators use this one-task form to stop their local pane
// before publishing the release event; the batch helper remains for
// deployments without a local coordinator.
func (s *Store) ExpireLease(id string, now time.Time) (domain.Event, error) {
	t, ok := s.Task(id)
	if !ok {
		return domain.Event{}, domain.ErrNotFound
	}
	if (t.State != domain.StateLeased && t.State != domain.StateNeedsAttention) || t.Lease == nil || t.Lease.Until.After(now) {
		return domain.Event{}, domain.ErrConflict
	}
	p, _ := json.Marshal(map[string]any{"reason": "lease_expired", "harness_id": t.Lease.HarnessID, "lease_epoch": t.Lease.Epoch, "expected_version": t.Version})
	e := domain.Event{ID: domain.NewID(), Type: "TaskReleased", TaskID: t.ID, Version: t.Version + 1, Payload: p, Surface: string(authz.System)}
	return e, s.Append(e)
}