// Package nativememory owns Desktop's durable local-first Memory checkpoint
// journal. It never knows about AGS or Git remotes: local Git is used only as
// an immutable on-device version journal, while Remote publishes bounded full
// trees through the native Memory API.
package nativememory

import (
	"context"
	"crypto/rand"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"os"
	pathpkg "path"
	"path/filepath"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/go-git/go-git/v5"
	"github.com/go-git/go-git/v5/plumbing"
	"github.com/go-git/go-git/v5/plumbing/filemode"
	"github.com/go-git/go-git/v5/plumbing/object"
)

const (
	stateSchemaVersion = 1
	maxFiles           = 2_000
	maxFileBytes       = int64(16 << 20)
	maxSnapshotBytes   = int64(64 << 20)
	defaultMinInterval = 5 * time.Minute
)

var (
	ErrNothingPending    = errors.New("native memory: nothing pending")
	ErrNotEligible       = errors.New("native memory: checkpoint is not eligible yet")
	ErrRemoteHeadChanged = errors.New("native memory: cloud head changed while local work is pending")
)

type Binding struct {
	MemoryID             string
	ContentID            string
	ContentVersionID     string
	Generation           uint64
	AcknowledgedSequence uint64
	ReplicaID            string
}

type File struct {
	Path       string
	Content    []byte
	Executable bool
}

type Checkpoint struct {
	OperationID            string
	ReplicaID              string
	ExpectedVersionID      string
	Generation             uint64
	CoveredFromSequence    uint64
	CoveredThroughSequence uint64
	Files                  []File
}

type Result struct {
	VersionID              string
	OperationID            string
	ReplicaID              string
	Generation             uint64
	CoveredFromSequence    uint64
	CoveredThroughSequence uint64
}

type Remote interface {
	PublishCheckpoint(context.Context, string, Checkpoint) (Result, error)
	OpenVersionArchive(context.Context, string, string) (io.ReadCloser, error)
}

type Options struct {
	Root        string
	StatePath   string
	Binding     Binding
	Remote      Remote
	Now         func() time.Time
	MinInterval time.Duration
	// persistState is a package-private durability seam used by failure-path
	// tests. Production always uses the atomic sidecar writer.
	persistState func(string, state) error
}

type pendingCheckpoint struct {
	OperationID string `json:"operationId"`
	From        uint64 `json:"from"`
	Through     uint64 `json:"through"`
	Revision    string `json:"revision"`
}

type state struct {
	SchemaVersion        int                `json:"schemaVersion"`
	MemoryID             string             `json:"memoryId"`
	ContentID            string             `json:"contentId"`
	Generation           uint64             `json:"generation"`
	ReplicaID            string             `json:"replicaId"`
	RemoteVersionID      string             `json:"remoteVersionId"`
	AcknowledgedSequence uint64             `json:"acknowledgedSequence"`
	LocalSequence        uint64             `json:"localSequence"`
	LocalRevision        string             `json:"localRevision"`
	Pending              *pendingCheckpoint `json:"pending,omitempty"`
	RefreshingVersionID  string             `json:"refreshingVersionId,omitempty"`
	LastAttemptAt        time.Time          `json:"lastAttemptAt,omitempty"`
}

type Sync struct {
	mu                      sync.Mutex
	root                    string
	statePath               string
	remote                  Remote
	now                     func() time.Time
	minInterval             time.Duration
	observedRemoteVersionID string
	state                   state
	persistState            func(string, state) error
}

func Open(opts Options) (*Sync, error) {
	root := strings.TrimSpace(opts.Root)
	statePath := strings.TrimSpace(opts.StatePath)
	if root == "" || statePath == "" || opts.Remote == nil {
		return nil, errors.New("native memory: root, state path, and remote are required")
	}
	var persisted *state
	if data, readErr := os.ReadFile(statePath); readErr == nil {
		var decoded state
		if err := json.Unmarshal(data, &decoded); err != nil {
			return nil, fmt.Errorf("native memory: decode state: %w", err)
		}
		persisted = &decoded
	} else if !os.IsNotExist(readErr) {
		return nil, fmt.Errorf("native memory: read state: %w", readErr)
	}
	binding := opts.Binding
	if strings.TrimSpace(binding.ReplicaID) == "" {
		if persisted != nil {
			binding.ReplicaID = persisted.ReplicaID
		} else {
			generated, err := newReplicaID()
			if err != nil {
				return nil, err
			}
			binding.ReplicaID = generated
		}
	}
	binding, err := normalizeBinding(binding)
	if err != nil {
		return nil, err
	}
	now := opts.Now
	if now == nil {
		now = time.Now
	}
	interval := opts.MinInterval
	if interval <= 0 {
		interval = defaultMinInterval
	}
	current := state{
		SchemaVersion: stateSchemaVersion, MemoryID: binding.MemoryID, ContentID: binding.ContentID,
		Generation: binding.Generation, ReplicaID: binding.ReplicaID,
		RemoteVersionID: binding.ContentVersionID, AcknowledgedSequence: binding.AcknowledgedSequence,
		LocalSequence: binding.AcknowledgedSequence,
	}
	if persisted != nil {
		current = *persisted
		if err := validateState(current, binding); err != nil {
			return nil, err
		}
		if binding.AcknowledgedSequence > current.AcknowledgedSequence {
			if binding.AcknowledgedSequence > current.LocalSequence || current.Pending == nil || current.Pending.Through != binding.AcknowledgedSequence {
				return nil, errors.New("native memory: server acknowledgement does not match the pending local operation")
			}
			// Replica/sequence proves that the frozen operation committed, but the
			// current binding head may already be a later restore. Advance only the
			// acknowledgement cursor here; active-head admission remains below.
			current.AcknowledgedSequence = binding.AcknowledgedSequence
			current.Pending = nil
		}
		if current.RemoteVersionID != binding.ContentVersionID {
			if current.Pending != nil || current.LocalSequence > current.AcknowledgedSequence {
				// Persist a newly observed exact acknowledgement before returning the
				// head conflict. Restart must not retry an already committed operation.
				s := &Sync{root: root, statePath: statePath, remote: opts.Remote, now: now, minInterval: interval, state: current, persistState: opts.persistState}
				if s.persistState == nil {
					s.persistState = writeState
				}
				if err := s.persistLocked(current); err != nil {
					return nil, err
				}
				return nil, ErrRemoteHeadChanged
			}
			if current.RefreshingVersionID != "" && current.RefreshingVersionID != binding.ContentVersionID {
				return nil, errors.New("native memory: cloud head changed during an unfinished local refresh")
			}
		}
	}
	observedRemoteVersionID := ""
	if current.RemoteVersionID != binding.ContentVersionID {
		observedRemoteVersionID = binding.ContentVersionID
	}
	s := &Sync{root: root, statePath: statePath, remote: opts.Remote, now: now, minInterval: interval, observedRemoteVersionID: observedRemoteVersionID, state: current, persistState: opts.persistState}
	if s.persistState == nil {
		s.persistState = writeState
	}
	if err := s.persistLocked(current); err != nil {
		return nil, err
	}
	return s, nil
}

func newReplicaID() (string, error) {
	var value [16]byte
	if _, err := rand.Read(value[:]); err != nil {
		return "", fmt.Errorf("native memory: generate replica id: %w", err)
	}
	return "desktop-" + hex.EncodeToString(value[:]), nil
}

func normalizeBinding(binding Binding) (Binding, error) {
	binding.MemoryID = strings.TrimSpace(binding.MemoryID)
	binding.ContentID = strings.TrimSpace(binding.ContentID)
	binding.ContentVersionID = strings.TrimSpace(binding.ContentVersionID)
	binding.ReplicaID = strings.TrimSpace(binding.ReplicaID)
	if binding.MemoryID == "" || binding.ContentID == "" || binding.ContentVersionID == "" || binding.Generation == 0 || binding.ReplicaID == "" {
		return Binding{}, errors.New("native memory: incomplete binding")
	}
	return binding, nil
}

func validateState(current state, binding Binding) error {
	if current.SchemaVersion != stateSchemaVersion {
		return fmt.Errorf("native memory: unsupported state schema %d", current.SchemaVersion)
	}
	if current.MemoryID != binding.MemoryID || current.ContentID != binding.ContentID || current.Generation != binding.Generation || current.ReplicaID != binding.ReplicaID {
		return errors.New("native memory: persisted journal belongs to another binding generation")
	}
	if current.AcknowledgedSequence > current.LocalSequence || current.AcknowledgedSequence > binding.AcknowledgedSequence {
		return errors.New("native memory: invalid acknowledged sequence")
	}
	if current.Pending != nil && (current.Pending.From != current.AcknowledgedSequence+1 || current.Pending.Through < current.Pending.From || current.Pending.Through > current.LocalSequence || strings.TrimSpace(current.Pending.Revision) == "") {
		return errors.New("native memory: invalid pending checkpoint")
	}
	if current.RefreshingVersionID != "" && (current.Pending != nil || current.LocalSequence != current.AcknowledgedSequence) {
		return errors.New("native memory: invalid remote refresh state")
	}
	return nil
}

// RecordLocalRevision durably advances the local journal after a local Git
// commit. Replaying the same revision is idempotent. A frozen pending attempt
// remains unchanged so retries always reuse identical inputs.
func (s *Sync) RecordLocalRevision(revision string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	revision = strings.TrimSpace(revision)
	if !validRevision(revision) {
		return errors.New("native memory: invalid local revision")
	}
	if s.state.RefreshingVersionID != "" {
		return errors.New("native memory: cannot record local work during a cloud refresh")
	}
	if revision == s.state.LocalRevision {
		return nil
	}
	next := s.state
	next.LocalSequence++
	next.LocalRevision = revision
	return s.persistLocked(next)
}

// AdoptRemoteRevision records the locally materialized exact cloud head as the
// journal baseline without creating a new unsynced local sequence.
func (s *Sync) AdoptRemoteRevision(revision string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	revision = strings.TrimSpace(revision)
	if !validRevision(revision) {
		return errors.New("native memory: invalid remote revision")
	}
	if s.state.LocalRevision == revision {
		return nil
	}
	if s.state.LocalRevision != "" || s.state.Pending != nil || s.state.LocalSequence != s.state.AcknowledgedSequence {
		return errors.New("native memory: cannot replace an established local journal baseline")
	}
	next := s.state
	next.LocalRevision = revision
	return s.persistLocked(next)
}

// RemoteRefreshVersion returns the exact newer cloud version observed during
// Open. An empty value means the local baseline already matches the server.
func (s *Sync) RemoteRefreshVersion() string {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.observedRemoteVersionID
}

// RemoteVersionID returns the exact cloud head that is already durable in the
// sidecar. Active sessions use this instead of an attach-time snapshot because
// a successful local publish advances the head without rebuilding the watcher.
func (s *Sync) RemoteVersionID() string {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.state.RemoteVersionID
}

// LocalRevision returns the exact immutable journal head recorded in the
// durable sidecar.
func (s *Sync) LocalRevision() string {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.state.LocalRevision
}

func (s *Sync) RemoteRefreshStarted() bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.state.RefreshingVersionID != ""
}

// BeginRemoteRefresh durably marks an exact cloud-head materialization before
// the working tree changes. Restart can therefore resume the same replacement
// without mistaking partial restore writes for new local edits.
func (s *Sync) BeginRemoteRefresh(versionID string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	versionID = strings.TrimSpace(versionID)
	if versionID == "" || versionID != s.observedRemoteVersionID {
		return errors.New("native memory: remote refresh does not match the observed cloud head")
	}
	if s.state.Pending != nil || s.state.LocalSequence != s.state.AcknowledgedSequence {
		return ErrRemoteHeadChanged
	}
	if s.state.RefreshingVersionID != "" && s.state.RefreshingVersionID != versionID {
		return errors.New("native memory: another cloud refresh is already in progress")
	}
	next := s.state
	next.RefreshingVersionID = versionID
	return s.persistLocked(next)
}

// AbortRemoteRefresh returns an admitted-but-unpublished cloud refresh to its
// prior durable baseline. It is used when new local work arrives after an
// isolated staged refresh failed without changing the live journal.
func (s *Sync) AbortRemoteRefresh(versionID string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	versionID = strings.TrimSpace(versionID)
	if versionID == "" || versionID != s.observedRemoteVersionID || s.state.RefreshingVersionID != versionID {
		return errors.New("native memory: aborted cloud refresh does not match the durable refresh state")
	}
	next := s.state
	next.RefreshingVersionID = ""
	return s.persistLocked(next)
}

// CompleteRemoteRefresh adopts a locally committed exact cloud tree as the
// new baseline without creating an outbound sequence. The prior tree remains
// in local Git history as the preceding immutable local version.
func (s *Sync) CompleteRemoteRefresh(versionID, revision string) error {
	s.mu.Lock()
	defer s.mu.Unlock()
	versionID = strings.TrimSpace(versionID)
	revision = strings.TrimSpace(revision)
	if versionID == "" || versionID != s.observedRemoteVersionID || s.state.RefreshingVersionID != versionID || !validRevision(revision) {
		return errors.New("native memory: completed cloud refresh does not match the durable refresh state")
	}
	next := s.state
	next.RemoteVersionID = versionID
	next.LocalRevision = revision
	next.RefreshingVersionID = ""
	next.LastAttemptAt = time.Time{}
	if err := s.persistLocked(next); err != nil {
		return err
	}
	s.observedRemoteVersionID = ""
	return nil
}

func (s *Sync) Publish(ctx context.Context) (Result, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.state.RefreshingVersionID != "" {
		return Result{}, errors.New("native memory: cloud refresh is incomplete")
	}
	if s.state.LocalSequence <= s.state.AcknowledgedSequence || s.state.LocalRevision == "" {
		return Result{}, ErrNothingPending
	}
	now := s.now().UTC()
	if !s.state.LastAttemptAt.IsZero() && now.Before(s.state.LastAttemptAt.Add(s.minInterval)) {
		return Result{}, ErrNotEligible
	}
	if s.state.Pending == nil {
		pending := &pendingCheckpoint{
			From: s.state.AcknowledgedSequence + 1, Through: s.state.LocalSequence, Revision: s.state.LocalRevision,
		}
		pending.OperationID = checkpointOperationID(s.state, *pending)
		next := s.state
		next.Pending = pending
		if err := s.persistLocked(next); err != nil {
			return Result{}, err
		}
	}
	pending := *s.state.Pending
	files, err := filesAtRevision(s.root, pending.Revision)
	if err != nil {
		return Result{}, err
	}
	next := s.state
	next.LastAttemptAt = now
	if err := s.persistLocked(next); err != nil {
		return Result{}, err
	}
	input := Checkpoint{
		OperationID: pending.OperationID, ReplicaID: s.state.ReplicaID,
		ExpectedVersionID: s.state.RemoteVersionID, Generation: s.state.Generation,
		CoveredFromSequence: pending.From, CoveredThroughSequence: pending.Through, Files: files,
	}
	result, err := s.remote.PublishCheckpoint(ctx, s.state.MemoryID, input)
	if err != nil {
		return Result{}, err
	}
	if strings.TrimSpace(result.VersionID) == "" || result.OperationID != input.OperationID || result.ReplicaID != input.ReplicaID || result.Generation != input.Generation || result.CoveredFromSequence != input.CoveredFromSequence || result.CoveredThroughSequence != input.CoveredThroughSequence {
		return Result{}, errors.New("native memory: checkpoint acknowledgement does not match pending operation")
	}
	next = s.state
	next.RemoteVersionID = strings.TrimSpace(result.VersionID)
	next.AcknowledgedSequence = pending.Through
	next.Pending = nil
	if err := s.persistLocked(next); err != nil {
		return Result{}, err
	}
	return result, nil
}

func (s *Sync) Pending() bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.state.LocalSequence > s.state.AcknowledgedSequence
}

func checkpointOperationID(current state, pending pendingCheckpoint) string {
	digest := sha256.Sum256([]byte(fmt.Sprintf("%s\x00%s\x00%d\x00%d\x00%d\x00%s", current.ReplicaID, current.ContentID, current.Generation, pending.From, pending.Through, pending.Revision)))
	return "desktop-checkpoint:" + hex.EncodeToString(digest[:])
}

func validRevision(value string) bool {
	if len(value) != 40 {
		return false
	}
	_, err := hex.DecodeString(value)
	return err == nil
}

func filesAtRevision(root, revision string) ([]File, error) {
	repo, err := git.PlainOpen(strings.TrimSpace(root))
	if err != nil {
		return nil, fmt.Errorf("native memory: open local journal: %w", err)
	}
	commit, err := repo.CommitObject(plumbing.NewHash(revision))
	if err != nil {
		return nil, fmt.Errorf("native memory: open local revision: %w", err)
	}
	tree, err := commit.Tree()
	if err != nil {
		return nil, fmt.Errorf("native memory: open local tree: %w", err)
	}
	files := make([]File, 0, 16)
	var total int64
	err = tree.Files().ForEach(func(item *object.File) error {
		path := filepath.ToSlash(item.Name)
		if path == "" || strings.TrimSpace(path) != path || strings.Contains(path, "\\") || pathpkg.Clean(path) != path || strings.HasPrefix(path, "/") || path == ".." || strings.HasPrefix(path, "../") {
			return fmt.Errorf("native memory: unsafe checkpoint path %q", item.Name)
		}
		if reservedSnapshotPath(path) {
			return nil
		}
		if len(files) >= maxFiles {
			return fmt.Errorf("native memory: checkpoint exceeds %d files", maxFiles)
		}
		if item.Mode != filemode.Regular && item.Mode != filemode.Executable {
			return fmt.Errorf("native memory: unsupported file mode for %q", path)
		}
		reader, err := item.Reader()
		if err != nil {
			return err
		}
		data, err := io.ReadAll(io.LimitReader(reader, maxFileBytes+1))
		closeErr := reader.Close()
		if err != nil {
			return err
		}
		if closeErr != nil {
			return closeErr
		}
		if int64(len(data)) > maxFileBytes || total > maxSnapshotBytes-int64(len(data)) {
			return errors.New("native memory: checkpoint exceeds byte limits")
		}
		total += int64(len(data))
		files = append(files, File{Path: path, Content: data, Executable: item.Mode == filemode.Executable})
		return nil
	})
	if err != nil {
		return nil, err
	}
	sort.Slice(files, func(i, j int) bool { return files[i].Path < files[j].Path })
	return files, nil
}

func reservedSnapshotPath(value string) bool {
	for _, segment := range strings.Split(filepath.ToSlash(value), "/") {
		if segment == ".git" || segment == ".derived" || strings.HasPrefix(segment, ".agentteam-") {
			return true
		}
	}
	return false
}

// persistLocked publishes next in memory only after the atomic writer has
// made it durable. A failed write therefore leaves every idempotency check
// observing the previous durable state.
func (s *Sync) persistLocked(next state) error {
	if err := s.persistState(s.statePath, next); err != nil {
		return err
	}
	s.state = next
	return nil
}

func writeState(statePath string, current state) error {
	if err := os.MkdirAll(filepath.Dir(statePath), 0o700); err != nil {
		return fmt.Errorf("native memory: mkdir state: %w", err)
	}
	data, err := json.MarshalIndent(current, "", "  ")
	if err != nil {
		return fmt.Errorf("native memory: encode state: %w", err)
	}
	data = append(data, '\n')
	tmp, err := os.CreateTemp(filepath.Dir(statePath), ".native-memory-state-*")
	if err != nil {
		return fmt.Errorf("native memory: create state temp: %w", err)
	}
	tmpPath := tmp.Name()
	defer os.Remove(tmpPath)
	if err := tmp.Chmod(0o600); err != nil {
		_ = tmp.Close()
		return err
	}
	if _, err := tmp.Write(data); err != nil {
		_ = tmp.Close()
		return err
	}
	if err := tmp.Sync(); err != nil {
		_ = tmp.Close()
		return err
	}
	if err := tmp.Close(); err != nil {
		return err
	}
	if err := os.Rename(tmpPath, statePath); err != nil {
		return fmt.Errorf("native memory: publish state: %w", err)
	}
	dir, err := os.Open(filepath.Dir(statePath))
	if err != nil {
		return err
	}
	err = dir.Sync()
	closeErr := dir.Close()
	if err != nil {
		return err
	}
	return closeErr
}