orchestra/internal/store/cursor.go

package store

import (
	"encoding/json"
	"fmt"
	"os"
	"path/filepath"
	"sort"
	"strings"

	"orchestra/internal/domain"
)

// SourceCursor is how far a task has been reconciled against one external
// human-input source. Its meaning belongs to the provider: a Gitea comment
// id, a Vikunja activity id, a web command sequence. Orchestra only requires
// that the provider can resume from it.
//
// The cursor is an efficiency bound, never the correctness guarantee. A
// cursor that fails to persist after a decision was appended must not create
// a second decision, so provenance uniqueness on (provider, external_id) is
// what actually prevents duplicates. See Store.DecisionForSource.
type SourceCursor struct {
	TaskID   string `json:"task_id"`
	Provider string `json:"provider"`
	Cursor   string `json:"cursor"`
}

func cursorKey(taskID, provider string) string { return taskID + "\x00" + provider }

func (s *Store) SourceCursor(taskID, provider string) (SourceCursor, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	v, ok := s.cursors[cursorKey(taskID, provider)]
	if !ok {
		return SourceCursor{TaskID: taskID, Provider: provider}, false
	}
	return SourceCursor{TaskID: taskID, Provider: provider, Cursor: v}, true
}

// SetSourceCursor persists the cursor before returning. A caller must only
// advance it after every event it derived from that input is durable.
func (s *Store) SetSourceCursor(c SourceCursor) error {
	if strings.TrimSpace(c.TaskID) == "" || strings.TrimSpace(c.Provider) == "" {
		return fmt.Errorf("%w: cursor needs task_id and provider", domain.ErrInvalid)
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	prior, had := s.cursors[cursorKey(c.TaskID, c.Provider)]
	s.cursors[cursorKey(c.TaskID, c.Provider)] = c.Cursor
	if err := s.writeCursorsLocked(); err != nil {
		if had {
			s.cursors[cursorKey(c.TaskID, c.Provider)] = prior
		} else {
			delete(s.cursors, cursorKey(c.TaskID, c.Provider))
		}
		return err
	}
	return nil
}

// DecisionForSource resolves the decision already recorded for one external
// human utterance, so a refetch after a lost cursor write is a skip rather
// than a second decision.
func (s *Store) DecisionForSource(provider, externalID string) (string, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	id, ok := s.decisionSource[provider+"\x00"+externalID]
	return id, ok
}

func (s *Store) writeCursorsLocked() error {
	keys := make([]string, 0, len(s.cursors))
	for k := range s.cursors {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	out := make([]SourceCursor, 0, len(keys))
	for _, k := range keys {
		task, provider, _ := strings.Cut(k, "\x00")
		out = append(out, SourceCursor{TaskID: task, Provider: provider, Cursor: s.cursors[k]})
	}
	b, err := json.Marshal(out)
	if err != nil {
		return err
	}
	tmp := s.cursorPath + ".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.cursorPath); err != nil {
		return err
	}
	dir, err := os.Open(filepath.Dir(s.cursorPath))
	if err != nil {
		return err
	}
	defer dir.Close()
	return dir.Sync()
}

func (s *Store) loadCursors() error {
	b, err := os.ReadFile(s.cursorPath)
	if os.IsNotExist(err) {
		return nil
	}
	if err != nil {
		return err
	}
	var in []SourceCursor
	if err := json.Unmarshal(b, &in); err != nil {
		return fmt.Errorf("source cursors: %w", err)
	}
	for _, c := range in {
		s.cursors[cursorKey(c.TaskID, c.Provider)] = c.Cursor
	}
	return nil
}