package console

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"net/http"
	"os"
	"strconv"
	"time"

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

const runReadFailedMsg = "could not read the run. try again"

// runCall is one follow route's resolved request.
type runCall struct {
	p     Principal
	actor string
	acct  *Account
	app   string
	host  string
	id    string
}

func noRunMsg(host, id string) string { return fmt.Sprintf("%s has no run %s", host, id) }

// runGate is the four follow routes' shared gate, in the create door's
// order: a person or a run service, a member (the byte-identical 404), the
// host-claim refusal, the runs budget, then a run id of the right shape. A
// service then reaches only a run its own credential started: any other is
// answered exactly as an unknown id is, so a run's existence does not leak.
func (s *Server) runGate(w http.ResponseWriter, r *http.Request) (runCall, bool) {
	p, ok := principalFrom(r.Context())
	if !ok || !runCaller(p) {
		http.NotFound(w, r)
		return runCall{}, false
	}
	actor, acct, app, ok := s.apiAccountApp(w, r)
	if !ok {
		return runCall{}, false
	}
	if s.cfg.HostClaim {
		jsonErrorCode(w, http.StatusConflict, runCodeHostClaim, claimRunUnsupportedMsg)
		return runCall{}, false
	}
	if !s.cli.runs.allow(hashToken(bearer(r)), time.Now()) {
		jsonErrorCode(w, http.StatusTooManyRequests, runCodeRateLimited, runTooManyMsg)
		return runCall{}, false
	}
	c := runCall{p: p, actor: actor, acct: acct, app: app, host: r.PathValue("host"), id: r.PathValue("id")}
	if s.cfg.Releases == nil {
		jsonErrorCode(w, http.StatusServiceUnavailable, runCodeUnavailable, runReadFailedMsg)
		return runCall{}, false
	}
	if !provision.ValidHostShape(c.host) || !runs.ValidID(c.id) {
		jsonErrorCode(w, http.StatusNotFound, runCodeNotFound, noRunMsg(c.host, c.id))
		return runCall{}, false
	}
	if p.Kind == PrincipalService {
		req, err := runs.ReadRequest(r.Context(), s.cfg.Releases, acct.Sqid, app, c.id)
		switch {
		case err != nil && !errors.Is(err, os.ErrNotExist):
			jsonErrorCode(w, http.StatusBadGateway, runCodeReadFailed, runReadFailedMsg)
			return runCall{}, false
		case err != nil || !s.runVisibleTo(p, *req):
			jsonErrorCode(w, http.StatusNotFound, runCodeNotFound, noRunMsg(c.host, c.id))
			return runCall{}, false
		}
	}
	return c, true
}

// runOf is the record's run when it is c.id, nil otherwise (the record is
// gone, or names another run). A read failure is an error.
func (s *Server) runOf(ctx context.Context, c runCall) (*provision.RunState, error) {
	rec, err := provision.GetRecord(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.host)
	if errors.Is(err, os.ErrNotExist) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	if rec.ActiveRun != nil && rec.ActiveRun.ID == c.id {
		return rec.ActiveRun, nil
	}
	return nil, nil
}

// requestOnHost reports whether the run's objects exist and belong to c.host.
// A run id names one run in the app's subtree; the host in the path must be
// the host the run was admitted for. A read failure is an error.
func (s *Server) requestOnHost(ctx context.Context, c runCall) (bool, error) {
	req, err := runs.ReadRequest(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.id)
	if errors.Is(err, os.ErrNotExist) {
		return false, nil
	}
	if err != nil {
		return false, err
	}
	return req.Host == c.host, nil
}

// runWriteCtx is the context for a record write that must run to its end
// even if the caller has gone: a cancelled conditional write may or may not
// have landed, and what follows (an audit row, a delete) depends on which.
func runWriteCtx(ctx context.Context) (context.Context, context.CancelFunc) {
	return context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second)
}

// handleAPIRunGet is GET …/runs/{id}?after=<seq> (run spec §5.6). It reads
// result.json BEFORE it lists chunks: the box writes the result only after
// every chunk is durable, so a page that carries the result also carries
// (or pages on to) every chunk up to its last_seq. Reading them the other
// way round could list the chunks, then see a result written after, and
// tell the caller the run is complete with chunks missing.
func (s *Server) handleAPIRunGet(w http.ResponseWriter, r *http.Request) {
	c, ok := s.runGate(w, r)
	if !ok {
		return
	}
	ctx := r.Context()
	var after int64
	if q := r.URL.Query().Get("after"); q != "" {
		var err error
		if after, err = strconv.ParseInt(q, 10, 64); err != nil || after < 0 {
			jsonErrorCode(w, http.StatusBadRequest, runCodeMalformed, runMalformedMsg)
			return
		}
	}
	run, err := s.runOf(ctx, c)
	if err != nil {
		jsonErrorCode(w, http.StatusBadGateway, runCodeReadFailed, runReadFailedMsg)
		return
	}
	if run == nil {
		// The record no longer names the run (cleared, or another run since):
		// its objects are what is left of it.
		found, err := s.requestOnHost(ctx, c)
		switch {
		case err != nil:
			jsonErrorCode(w, http.StatusBadGateway, runCodeReadFailed, runReadFailedMsg)
			return
		case !found:
			jsonErrorCode(w, http.StatusNotFound, runCodeNotFound, noRunMsg(c.host, c.id))
			return
		}
	}
	res, err := runs.ReadResult(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.id)
	if err != nil {
		jsonErrorCode(w, http.StatusBadGateway, runCodeReadFailed, runReadFailedMsg)
		return
	}
	keys, err := runs.ListChunks(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.id, after, runs.PageChunks+1)
	if err != nil {
		jsonErrorCode(w, http.StatusBadGateway, runCodeReadFailed, runReadFailedMsg)
		return
	}
	p := runs.Page{ID: c.id, Result: res, Chunks: []runs.Chunk{}}
	if run != nil {
		p.State = run.State
	}
	out := runs.OutPrefix(c.acct.Sqid, c.app, c.id)
	size := 0
	for i, k := range keys {
		if i == runs.PageChunks {
			p.More = true
			break
		}
		seq, stream, ok := runs.ParseChunkKey(k, out)
		if !ok {
			continue
		}
		raw, _, err := s.cfg.Releases.Get(ctx, k)
		if err != nil {
			jsonErrorCode(w, http.StatusBadGateway, runCodeReadFailed, runReadFailedMsg)
			return
		}
		if size+len(raw) > runs.PageBytes && len(p.Chunks) > 0 {
			p.More = true
			break
		}
		size += len(raw)
		p.Chunks = append(p.Chunks, runs.Chunk{Seq: seq, Stream: stream, Data: raw})
	}
	if res != nil {
		s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, runEndOutcome(res), res)
	}
	writeJSON(w, http.StatusOK, p)
}

// handleAPIRunCancel is POST …/runs/{id}/cancel (run spec §5.7): queued
// becomes cancelled (terminal; the run-end row is written now), picked
// becomes cancel and the serial is bumped so the owning box hears it.
func (s *Server) handleAPIRunCancel(w http.ResponseWriter, r *http.Request) {
	c, ok := s.runGate(w, r)
	if !ok {
		return
	}
	ctx, cancel := runWriteCtx(r.Context())
	defer cancel()
	got, err := provision.SetRecordRun(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.host, provision.RunCancelTransition(c.id))
	if err != nil {
		s.runTransitionRefused(w, ctx, c, err)
		return
	}
	switch got.ActiveRun.State {
	case provision.RunCancelled:
		if err := s.endUnpicked(ctx, c, provision.RunCancelled); err != nil {
			log.Printf("console: api run %s/%s/%s %s: end after cancel: %v", c.acct.Sqid, c.app, c.host, c.id, err)
			jsonErrorCode(w, http.StatusBadGateway, runCodeRecordFailed, runRecordFailedMsg)
			return
		}
	case provision.RunCancel:
		s.bumpSerial(ctx)
	}
	writeJSON(w, http.StatusOK, runs.StateAnswer{State: got.ActiveRun.State})
}

// handleAPIRunExpire is POST …/runs/{id}/expire (run spec §5.2): the CLI's
// "not picked up", asked for sixty seconds after its request. The route
// applies the transition and does not judge the clock; the sweep, which
// expires on its own, decides by the record's pickup Deadline (stamped at
// admission, after the upload), never by request.json's requested_at. A
// pickup that landed first wins.
func (s *Server) handleAPIRunExpire(w http.ResponseWriter, r *http.Request) {
	c, ok := s.runGate(w, r)
	if !ok {
		return
	}
	ctx, cancel := runWriteCtx(r.Context())
	defer cancel()
	got, err := provision.SetRecordRun(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.host, provision.RunExpire(c.id))
	if err != nil {
		s.runTransitionRefused(w, ctx, c, err)
		return
	}
	if err := s.endUnpicked(ctx, c, provision.RunExpired); err != nil {
		log.Printf("console: api run %s/%s/%s %s: end after expire: %v", c.acct.Sqid, c.app, c.host, c.id, err)
		jsonErrorCode(w, http.StatusBadGateway, runCodeRecordFailed, runRecordFailedMsg)
		return
	}
	writeJSON(w, http.StatusOK, runs.StateAnswer{State: got.ActiveRun.State})
}

// endUnpicked finishes a run that ended before any box picked it: cancelled
// or expired is terminal, so no box will ever write its result.json. The
// console writes one (create-only, so a repeat changes nothing), so that every
// terminal run has a result a follower can read after the record has moved
// on, and then the run-end row. state is the record's terminal state.
func (s *Server) endUnpicked(ctx context.Context, c runCall, state string) error {
	res := runs.Result{Ended: time.Now().UTC()}
	if state == provision.RunCancelled {
		res.Cancelled = true
	} else {
		res.Refused = runs.RefusedExpired
	}
	if err := runs.WriteResult(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.id, res); err != nil {
		return err
	}
	s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, runEndOutcome(&res), &res)
	return nil
}

// runTransitionRefused names why a cancel or expire did not apply, from a
// fresh read of the record. A run the record shows cancelled or expired
// gets its result and run-end row here too (both written once, so a no-op
// when the call that ended it already wrote them): a cancel whose response
// was lost and was retried finds its own terminal state, and these must not
// depend on the first attempt having got that far.
func (s *Server) runTransitionRefused(w http.ResponseWriter, ctx context.Context, c runCall, err error) {
	if !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) {
		log.Printf("console: api run %s/%s/%s %s: %v", c.acct.Sqid, c.app, c.host, c.id, err)
		jsonErrorCode(w, http.StatusBadGateway, runCodeRecordFailed, runRecordFailedMsg)
		return
	}
	run, rerr := s.runOf(ctx, c)
	switch {
	case rerr != nil:
		jsonErrorCode(w, http.StatusBadGateway, runCodeReadFailed, runReadFailedMsg)
	case run == nil:
		jsonErrorCode(w, http.StatusNotFound, runCodeNotFound, noRunMsg(c.host, c.id))
	case run.Open():
		jsonErrorCode(w, http.StatusConflict, runCodeRunStarted, fmt.Sprintf("run %s has already started", c.id))
	default:
		if run.State == provision.RunCancelled || run.State == provision.RunExpired {
			if err := s.endUnpicked(ctx, c, run.State); err != nil {
				log.Printf("console: api run %s/%s/%s %s: end after retry: %v", c.acct.Sqid, c.app, c.host, c.id, err)
				jsonErrorCode(w, http.StatusBadGateway, runCodeRecordFailed, runRecordFailedMsg)
				return
			}
		}
		jsonErrorCode(w, http.StatusConflict, runCodeRunEnded, fmt.Sprintf("run %s has already ended", c.id))
	}
}

// handleAPIRunAck is POST …/runs/{id}/ack (run spec §5.6, §5.8, codex 5):
// refused unless the result exists and last_seq is the result's. Then the
// record's run is cleared and the run's objects are deleted, in that order,
// and only for a run the record does not hold in a live state.
func (s *Server) handleAPIRunAck(w http.ResponseWriter, r *http.Request) {
	c, ok := s.runGate(w, r)
	if !ok {
		return
	}
	var ack runs.Ack
	if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4<<10)).Decode(&ack); err != nil {
		jsonErrorCode(w, http.StatusBadRequest, runCodeMalformed, runMalformedMsg)
		return
	}
	ctx, cancel := runWriteCtx(r.Context())
	defer cancel()
	found, err := s.requestOnHost(ctx, c)
	switch {
	case err != nil:
		jsonErrorCode(w, http.StatusBadGateway, runCodeReadFailed, runReadFailedMsg)
		return
	case !found:
		jsonErrorCode(w, http.StatusNotFound, runCodeNotFound, noRunMsg(c.host, c.id))
		return
	}
	res, err := runs.ReadResult(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.id)
	switch {
	case err != nil:
		jsonErrorCode(w, http.StatusBadGateway, runCodeReadFailed, runReadFailedMsg)
		return
	case res == nil:
		jsonErrorCode(w, http.StatusConflict, runCodeRunNotFinished, fmt.Sprintf("run %s has not finished yet", c.id))
		return
	case res.LastSeq != ack.LastSeq:
		jsonErrorCode(w, http.StatusConflict, runCodeLastSeqMismatch, fmt.Sprintf("run %s ended at chunk %d, not %d", c.id, res.LastSeq, ack.LastSeq))
		return
	}
	run, err := s.runOf(ctx, c)
	switch {
	case err != nil:
		jsonErrorCode(w, http.StatusBadGateway, runCodeReadFailed, runReadFailedMsg)
		return
	case run.Open():
		// The box has written the result and not yet moved the run to done.
		// Its objects are the box's until it has.
		jsonErrorCode(w, http.StatusConflict, runCodeRunNotFinished, fmt.Sprintf("run %s has not finished yet", c.id))
		return
	}
	s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, runEndOutcome(res), res)
	if run != nil {
		// Clear the record first. If the CAS fails, or its outcome is unknown,
		// the record is asked again before anything is deleted: a run the
		// record still names keeps its objects, and the caller may ack again.
		_, cerr := provision.SetRecordRun(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.host, provision.RunClear(c.id))
		still, rerr := s.runOf(ctx, c)
		if rerr != nil || still != nil {
			log.Printf("console: api run %s/%s/%s %s: clear after ack: %v; record read: %v", c.acct.Sqid, c.app, c.host, c.id, cerr, rerr)
			jsonErrorCode(w, http.StatusBadGateway, runCodeRecordFailed, runRecordFailedMsg)
			return
		}
	}
	if err := runs.DeleteRun(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.id); err != nil {
		// The sweep deletes what is left (Task 5); the ack itself stands.
		log.Printf("console: api run %s/%s/%s %s: delete after ack: %v", c.acct.Sqid, c.app, c.host, c.id, err)
	}
	writeJSON(w, http.StatusOK, runs.StateAnswer{State: "acknowledged"})
}

// runEndOutcome is the run-end row's outcome, derived from the result alone so
// that whoever writes the row first (a follower's poll, an ack, a cancel, an
// expiry, the sweep) writes the same word: lost, cancelled, expired, or result.
func runEndOutcome(res *runs.Result) string {
	switch {
	case res.Interrupted == runs.InterruptedLost:
		return "lost"
	case res.Cancelled:
		return provision.RunCancelled
	case res.Refused == runs.RefusedExpired:
		return provision.RunExpired
	}
	return "result"
}

// auditRunEnd writes the run-end row exactly once per run (run spec §6):
// whoever creates runs/<id>/audited first writes it, from the CLI's poll,
// a cancel, an expiry or the sweep. The actor is the run's caller. It runs
// to its end even if the caller has gone, so a cancelled request cannot
// leave the marker without the row's chance.
func (s *Server) auditRunEnd(ctx context.Context, acct *Account, app, host, id, outcome string, res *runs.Result) {
	ctx, cancel := runWriteCtx(ctx)
	defer cancel()
	if _, err := s.cfg.Releases.Put(ctx, runs.AuditedKey(acct.Sqid, app, id), []byte(outcome), ""); err != nil {
		if !errors.Is(err, release.ErrCASConflict) {
			log.Printf("console: api run %s/%s/%s %s: audit marker: %v", acct.Sqid, app, host, id, err)
		}
		return // written already, or the bucket refused: either way not twice
	}
	// The marker is the last object DeleteRun removes, after request.json. A
	// marker Put that succeeds for a run whose request.json is gone therefore
	// raced a finished ack: the row was written already. Take the marker back
	// and write nothing.
	by, requester := "", ""
	switch req, err := runs.ReadRequest(ctx, s.cfg.Releases, acct.Sqid, app, id); {
	case errors.Is(err, os.ErrNotExist):
		if derr := s.cfg.Releases.Delete(ctx, runs.AuditedKey(acct.Sqid, app, id)); derr != nil {
			log.Printf("console: api run %s/%s/%s %s: remove stray audit marker: %v", acct.Sqid, app, host, id, derr)
		}
		return
	case err == nil:
		by, requester = req.By, req.ServiceLine()
	}
	d := struct {
		ID      string       `json:"id"`
		Outcome string       `json:"outcome"`
		Result  *runs.Result `json:"result,omitempty"`
		// Requester is a service's run's line, as on the run-start row.
		Requester string `json:"requester,omitempty"`
	}{id, outcome, res, requester}
	raw, _ := json.Marshal(d)
	s.auditActivity(acct, app, by, Activity{Action: "run-end", TargetType: "instance", Target: host, Detail: string(raw)})
}