package host

import (
	"context"
	"errors"
	"fmt"
	"io"
	"sync"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/registry"
)

type filesSeedIncompleteError struct{ error }

type filesSeedPending struct {
	exclusion *activator.SeedExclusion
	wake      *ClaimWakeAssembly
}
type filesSeedExecutor struct {
	assembly *ClaimAssembly
	service  *activator.Service
	registry *registry.Registry
	source   *movesFilesSource
	mu       sync.Mutex
	running  bool
	pending  map[string]filesSeedPending
}

func (e *filesSeedExecutor) run(ctx context.Context, host string, out io.Writer) error {
	if e == nil || e.assembly == nil || e.service == nil || e.registry == nil || e.source == nil || out == nil {
		return hostbox.ErrUnsupported
	}
	e.mu.Lock()
	if e.running {
		e.mu.Unlock()
		return activator.ErrFenceHeld
	}
	if e.pending == nil {
		e.pending = map[string]filesSeedPending{}
	}
	e.running = true
	e.mu.Unlock()
	defer func() { e.mu.Lock(); e.running = false; e.mu.Unlock() }()
	a := e.assembly
	if a.Wake == nil || a.Wake.seedFactory == nil {
		return hostbox.ErrUnsupported
	}
	actor, err := a.Identity.Verify(ctx)
	if err != nil {
		return err
	}
	candidates, err := filesSeedCandidates(ctx, a.Claims, actor.Box, host)
	if err != nil {
		return err
	}
	var failures []error
	for _, o := range candidates {
		if err := ctx.Err(); err != nil {
			return filesSeedIncompleteError{errors.Join(append(failures, err)...)}
		}
		if err := e.host(ctx, o, out); err != nil {
			failures = append(failures, err)
			if _, writeErr := fmt.Fprintf(out, "Host %q: failed: %s.\n", o.Host, err); writeErr != nil {
				return errors.Join(err, writeErr)
			}
		}
	}
	if err := errors.Join(failures...); err != nil {
		return filesSeedIncompleteError{err}
	}
	return nil
}

func filesSeedRetryReport(out io.Writer, o hostclaim.Observed) error {
	if o.Claim == nil || o.Claim.Holder == nil {
		return nil
	}
	r := o.Claim.Holder.WakeRetry
	if r == nil {
		r = &hostclaim.WakeRetry{State: "none"}
	}
	_, err := fmt.Fprintf(out, "Host %q: wake retry episode %q, state %q, reserved %d of 5, remaining %d, retry after %q.\n", o.Host, r.EpisodeID, r.State, r.AttemptsReserved, 5-r.AttemptsReserved, r.NotBefore.Format(time.RFC3339))
	return err
}

func (e *filesSeedExecutor) host(ctx context.Context, o hostclaim.Observed, out io.Writer) (resultErr error) {
	host := o.Host
	a := e.assembly
	actor, err := a.Identity.Verify(ctx)
	if err != nil {
		return err
	}
	// Scope is decided from the claim before the registry is consulted.
	if o.Absent || o.Claim == nil {
		_, err = fmt.Fprintf(out, "Host %q: skipped; no claim.\n", o.Host)
		return err
	}
	if o.Claim.Holder == nil || o.Claim.Op.Kind != hostclaim.KindServing || o.Claim.Op.Exec != actor.Box || o.Claim.Op.ID != "" || o.Claim.Op.Phase != "" {
		_, err = fmt.Fprintf(out, "Host %q: skipped; claim state %q, executor %q.\n", o.Host, o.Claim.Op.Kind, o.Claim.Op.Exec)
		return err
	}
	row, err := e.registry.Get(o.Host)
	if err != nil {
		return err
	}
	if row == nil {
		return hostbox.ErrChanged
	}
	if row.Replica {
		_, err := fmt.Fprintf(out, "Host %q: skipped; replica files are not an authoritative source.\n", o.Host)
		return err
	}
	if row.Seed != "" || row.Backing != "" || row.Unit != "" || row.IsSidecar() || row.Tunnel != "" {
		return hostbox.ErrUnsupported
	}
	opened, err := e.source.Open(ctx, a.Claims, hostbox.RequestStore{Store: a.Data, Check: claimBoundIdentity(a.Identity.Verify, actor)}, actor)
	if err != nil {
		return err
	}
	files, ok := opened.(*movesFiles)
	if !ok {
		return hostbox.ErrUnsupported
	}
	if _, err = checkFilesSeedLayout(files, *row, claimEmptyBootstrap(o.Claim.Holder)); err != nil {
		return err
	}
	pending := e.pending[o.Host]
	retry := o.Claim.Holder.WakeRetry
	needsSettlement := pending.exclusion != nil || retry != nil && retry.State == "attempting"
	var token *activator.SeedExclusion
	release := func() error {
		if token == nil {
			return nil
		}
		err := token.Release()
		if err == nil {
			delete(e.pending, o.Host)
			token = nil
		}
		return err
	}
	defer func() {
		if token != nil && e.pending[o.Host].exclusion == nil {
			resultErr = errors.Join(resultErr, release())
		}
	}()
	acquire := func() error {
		var err error
		token, err = e.service.AcquireSeedExclusion(ctx, o.Host, pending.exclusion)
		if err != nil {
			return err
		}
		fresh, readErr := hostclaim.Read(ctx, a.Claims, host)
		if readErr == nil {
			o = fresh
		}
		return readErr
	}
	if needsSettlement {
		if err := acquire(); err != nil {
			return err
		}
		if pending.wake != nil {
			if err := pending.wake.Close(ctx); err != nil {
				return err
			}
		}
		settled, settleErr := settleFilesSeedRerun(ctx, *row, files, a.Wake.Retries, o, token)
		err = settleErr
		if err != nil {
			e.pending[o.Host] = filesSeedPending{exclusion: token, wake: pending.wake}
			return err
		}
		delete(e.pending, o.Host)
		o = settled
	}
	selected, err := inspectFilesSeed(ctx, a.Claims, files, o, actor, row)
	if err != nil {
		return err
	}
	switch selected.State {
	case "established":
		if err := release(); err != nil {
			return err
		}
		_, err = fmt.Fprintf(out, "Host %q: an established files manifest already exists; no seed was written.\n", o.Host)
		return err
	case "emptyBootstrap":
		_, err = fmt.Fprintf(out, "Host %q: no files seed needed; the first wake starts empty.\n", o.Host)
		return err
	case "replica":
		_, err = fmt.Fprintf(out, "Host %q: skipped; replica files are not an authoritative source.\n", o.Host)
		return err
	case "eligible":
	default:
		return hostbox.ErrUnsupported
	}
	if token == nil {
		if err := acquire(); err != nil {
			return err
		}
		selected, err = inspectFilesSeed(ctx, a.Claims, files, o, actor, row)
		if err != nil {
			return err
		}
		if selected.State != "eligible" {
			return hostbox.ErrChanged
		}
	}
	if err := filesSeedRetryReport(out, o); err != nil {
		return err
	}
	defer func() {
		reportCtx, cancel := activator.CleanupContext(ctx, hostbox.RequestLimit)
		defer cancel()
		fresh, err := hostclaim.Read(reportCtx, a.Claims, o.Host)
		if err == nil {
			err = filesSeedRetryReport(out, fresh)
		}
		resultErr = errors.Join(resultErr, err)
	}()
	retry = o.Claim.Holder.WakeRetry
	if retry != nil {
		if retry.State != "open" || retry.AttemptsReserved >= 5 {
			return hostclaim.ErrWrongState
		}
		delay := time.Until(retry.NotBefore)
		if delay > 0 {
			timer := time.NewTimer(delay)
			defer timer.Stop()
			select {
			case <-ctx.Done():
				return ctx.Err()
			case <-timer.C:
			}
		}
	}
	_, op, err := newFilesSeedProvider(ctx, e.source, a.Claims, files.Data, actor, e.registry, selected)
	if err != nil {
		return err
	}
	runtime, wake, err := a.Wake.seedFactory(ctx, op, token)
	if err != nil {
		return err
	}
	if _, err = selectFilesSeedInputs(ctx, wake.Inputs, *row, selected); err != nil {
		return err
	}
	admitted, settled, err := prepareFilesSeed(ctx, *row, runtime, o, token)
	err = errors.Join(err, op.reportSkipped(out, o.Host))
	if !settled {
		e.pending[o.Host] = filesSeedPending{exclusion: token, wake: wake}
	}
	if admitted.Claim != nil && admitted.Claim.Holder != nil {
		_, writeErr := fmt.Fprintf(out, "Host %q: admitted files prefix %q; retain this line for interrupted-run cleanup.\n", o.Host, movesFilesPrefix(admitted.Claim.Holder.Location, admitted.Claim.Holder.Generation))
		err = errors.Join(err, writeErr)
	}
	if err != nil {
		return err
	}
	_, err = fmt.Fprintf(out, "Host %q: seeded.\n", o.Host)
	return err
}

// close is called only after the serial handler has joined. A failed private
// runtime stays retained until it joins; shared owners must not close beneath it.
func (e *filesSeedExecutor) close(ctx context.Context) error {
	e.mu.Lock()
	defer e.mu.Unlock()
	if e.running {
		return activator.ErrFenceHeld
	}
	var result error
	for _, pending := range e.pending {
		if pending.wake != nil {
			result = errors.Join(result, pending.wake.Close(ctx))
		}
	}
	return result
}