package hostfiles

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"slices"
	"sort"
	"strings"
	"sync"

	"github.com/carlosframework/platform/internal/release"
)

// ErrSuperseded means another writer replaced the manifest. The caller
// stops publishing: it is no longer this host's executor, or two
// publishers were started for one host.
var ErrSuperseded = errors.New("hostfiles: another writer published this manifest")

// Publisher stages changes and commits them as one manifest revision. It
// is safe for concurrent use.
type Publisher struct {
	s      release.Store
	prefix string

	mu    sync.Mutex
	man   Manifest
	etag  string
	next  int64 // the last sequence number handed out; man.Seq is the last committed
	dirty bool
	// inflight holds, by sha256, every manifest body a failed Put may have
	// applied since the last successful commit, with the sequence number it
	// carried. Every one of those Puts named the same etag, so at most one
	// of them can be in storage.
	inflight map[string]int64
}

func Open(ctx context.Context, s release.Store, prefix string) (*Publisher, error) {
	if err := checkPrefix(prefix); err != nil {
		return nil, err
	}
	m, etag, _, err := load(ctx, s, prefix)
	if err != nil {
		return nil, err
	}
	// A manifest written before the path cap may hold a path past it. No
	// reader accepts it (checkManifest), and no Commit could ever write it
	// again, so it is refused here, terminally, rather than continued into
	// a publisher whose every commit fails. An operator replaces it.
	if err := manifestWithinCap(m); err != nil {
		return nil, fmt.Errorf("hostfiles: open %s: the manifest cannot be continued, an operator must replace it: %w", prefix, err)
	}
	return &Publisher{s: s, prefix: prefix, man: m, etag: etag, next: m.Seq}, nil
}

// RestorableWithinCap is the check a caller makes before it starts work
// that will need m restored (a move's cut, the claim's files backfill):
// every entry path of m, live, tombstoned or skipped, within the tree's
// length cap. A manifest that fails it is ErrTooDeep: no reader accepts
// it, so an operator replaces it before anything that needs it starts
// (#236, Paul's ruling of 2026-10-15).
func RestorableWithinCap(m Manifest) error { return manifestWithinCap(m) }

// manifestWithinCap holds every entry path of m to withinTreeCap, the cap
// the readers apply: Open refuses to continue a manifest that fails it,
// and Commit refuses to write one.
func manifestWithinCap(m Manifest) error {
	for path := range m.Entries {
		if err := withinTreeCap(path); err != nil {
			return err
		}
	}
	return nil
}

// OpenAt is a publisher over the manifest bytes a caller already read and
// checked, with the token the store answered for them: the publisher's
// baseline and its compare-and-swap token are exactly those, never a
// second read (Task 6 fix round 2, finding 2). A manifest replaced at the
// key after that read makes the first Commit fail (ErrSuperseded) rather
// than publish over bytes nobody checked.
func OpenAt(s release.Store, prefix string, body []byte, etag string) (*Publisher, error) {
	if s == nil {
		return nil, errors.New("hostfiles: open: no store")
	}
	if err := checkPrefix(prefix); err != nil {
		return nil, err
	}
	if etag == "" {
		return nil, fmt.Errorf("hostfiles: open %s at its read bytes: no token", prefix)
	}
	m, err := decode(prefix, body)
	if err != nil {
		return nil, err
	}
	// The same terminal refusal as Open's: a manifest past the path cap
	// cannot be continued.
	if err := manifestWithinCap(m); err != nil {
		return nil, fmt.Errorf("hostfiles: open %s: the manifest cannot be continued, an operator must replace it: %w", prefix, err)
	}
	return &Publisher{s: s, prefix: prefix, man: m, etag: etag, next: m.Seq}, nil
}

func sum(body []byte) string {
	h := sha256.Sum256(body)
	return hex.EncodeToString(h[:])
}

// put writes body as the next immutable object under this publisher's own
// prefix and returns its entry, whose Object is the full key. Callers hold
// p.mu.
func (p *Publisher) put(ctx context.Context, body []byte, mode uint32, torn bool) (Entry, error) {
	sha := sum(body)
	seq := p.next + 1
	key := p.prefix + objectKey(seq, sha)
	if _, err := p.s.Put(ctx, key, body, ""); err != nil && !errors.Is(err, release.ErrCASConflict) {
		// A conflict means this exact key exists: the same sequence number
		// and the same hash, left by an attempt that crashed before its
		// commit. The bytes are the ones we were about to write.
		return Entry{}, err
	}
	p.next = seq
	return Entry{Object: key, SHA256: sha, Size: int64(len(body)), Mode: mode & 0o777, Seq: seq, Torn: torn}, nil
}

// Stage records path's content. Content identical to what is already
// staged or committed takes no sequence number and writes nothing.
func (p *Publisher) Stage(ctx context.Context, path string, body []byte, mode uint32, torn bool) error {
	clean, err := CleanPath(path)
	if err != nil {
		return err
	}
	if err := withinTreeCap(clean); err != nil {
		return err
	}
	p.mu.Lock()
	defer p.mu.Unlock()
	if cur, ok := p.man.Entries[clean]; ok && !cur.Deleted && cur.Skipped == "" &&
		cur.SHA256 == sum(body) && cur.Mode == mode&0o777 && cur.Torn == torn {
		return nil
	}
	e, err := p.put(ctx, body, mode, torn)
	if err != nil {
		return err
	}
	p.man.Entries[clean], p.dirty = e, true
	return nil
}

// StageSkipped records that path exists but was not published, and why.
func (p *Publisher) StageSkipped(path, why string) error {
	clean, err := CleanPath(path)
	if err != nil {
		return err
	}
	if err := withinTreeCap(clean); err != nil {
		return err
	}
	p.mu.Lock()
	defer p.mu.Unlock()
	if cur, ok := p.man.Entries[clean]; ok && cur.Skipped == why {
		return nil
	}
	p.next++
	p.man.Entries[clean], p.dirty = Entry{Seq: p.next, Skipped: why}, true
	return nil
}

// StageDelete records that path is gone. Deleting a path the manifest never
// had, or already has as deleted, is a no-op.
func (p *Publisher) StageDelete(path string) error {
	clean, err := CleanPath(path)
	if err != nil {
		return err
	}
	if err := withinTreeCap(clean); err != nil {
		return err
	}
	p.mu.Lock()
	defer p.mu.Unlock()
	if cur, ok := p.man.Entries[clean]; !ok || cur.Deleted {
		return nil
	}
	p.next++
	p.man.Entries[clean], p.dirty = Entry{Seq: p.next, Deleted: true}, true
	return nil
}

// StagePlatform records a platform-managed item (wake state, an env file).
// It shares the manifest's ordering but is never written into the files
// directory by Restore.
func (p *Publisher) StagePlatform(ctx context.Context, name string, body []byte) error {
	if name == "" || strings.ContainsAny(name, "/\x00") {
		return fmt.Errorf("hostfiles: platform item %q is not a plain name", name)
	}
	p.mu.Lock()
	defer p.mu.Unlock()
	if cur, ok := p.man.Platform[name]; ok && cur.SHA256 == sum(body) {
		return nil
	}
	e, err := p.put(ctx, body, 0o600, false)
	if err != nil {
		return err
	}
	p.man.Platform[name], p.dirty = e, true
	return nil
}

// Commit publishes everything staged as one manifest revision and returns
// its sequence number. With nothing staged it returns the current one. A
// failed commit leaves the staged changes in place, so the caller retries;
// ErrSuperseded is final.
func (p *Publisher) Commit(ctx context.Context) (int64, error) {
	p.mu.Lock()
	defer p.mu.Unlock()
	if !p.dirty {
		return p.man.Seq, nil
	}
	// The readers refuse a manifest with a path past the cap
	// (checkManifest), so none is written: Open refuses to continue one and
	// Stage refuses to add one; this is the last line before the store.
	if err := manifestWithinCap(p.man); err != nil {
		return 0, err
	}
	out := p.man
	out.Seq = p.next
	body, err := json.Marshal(out)
	if err != nil {
		return 0, fmt.Errorf("hostfiles: marshal manifest: %w", err)
	}
	etag, err := p.s.Put(ctx, ManifestKey(p.prefix), body, p.etag)
	if errors.Is(err, release.ErrCASConflict) {
		// A lost response looks like this: the store applied an earlier Put,
		// the reply never arrived, and this retry's etag is now stale. If
		// the manifest holds exactly the bytes this Commit is writing, or
		// the bytes an earlier failed Commit wrote, it is ours.
		cur, curEtag, gerr := p.s.Get(ctx, ManifestKey(p.prefix))
		if gerr != nil {
			// Not knowing is not the same as being superseded: leave the
			// staged state and etag alone so the next Commit repeats the check.
			return 0, gerr
		}
		if bytes.Equal(cur, body) {
			err, etag = nil, curEtag
		} else if seq, ok := p.inflight[sum(cur)]; ok {
			// An earlier attempt landed and later stages changed the body.
			// Adopt it, then write the current body once on top of it.
			p.man.Seq, p.etag, p.inflight = seq, curEtag, nil
			etag, err = p.s.Put(ctx, ManifestKey(p.prefix), body, p.etag)
			if errors.Is(err, release.ErrCASConflict) {
				return 0, fmt.Errorf("%w: %s", ErrSuperseded, p.prefix)
			}
		} else {
			return 0, fmt.Errorf("%w: %s", ErrSuperseded, p.prefix)
		}
	}
	if err != nil {
		if !errors.Is(err, release.ErrCASConflict) {
			if p.inflight == nil {
				p.inflight = map[string]int64{}
			}
			p.inflight[sum(body)] = out.Seq
		}
		return 0, err
	}
	p.man.Seq, p.etag, p.dirty, p.inflight = out.Seq, etag, false, nil
	return out.Seq, nil
}

// SetOrigins sets the manifest's carried origin set (Manifest.Origins),
// sorted and deduplicated. The same set is not a change; a different one
// goes out with the next Commit.
func (p *Publisher) SetOrigins(origins []string) {
	set := map[string]bool{}
	for _, o := range origins {
		if o != "" {
			set[o] = true
		}
	}
	out := make([]string, 0, len(set))
	for o := range set {
		out = append(out, o)
	}
	sort.Strings(out)
	p.mu.Lock()
	defer p.mu.Unlock()
	if slices.Equal(out, p.man.Origins) {
		return
	}
	p.man.Origins, p.dirty = out, true
}

func (p *Publisher) Dirty() bool { p.mu.Lock(); defer p.mu.Unlock(); return p.dirty }

func (p *Publisher) Seq() int64 { p.mu.Lock(); defer p.mu.Unlock(); return p.man.Seq }

// Paths lists every path the manifest holds as present: published or
// skipped, not deleted.
func (p *Publisher) Paths() []string {
	p.mu.Lock()
	defer p.mu.Unlock()
	var out []string
	for path, e := range p.man.Entries {
		if !e.Deleted {
			out = append(out, path)
		}
	}
	sort.Strings(out)
	return out
}

// pending is the last sequence number handed out, committed or not.
func (p *Publisher) pending() int64 {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.next
}