package hostbox

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"time"
)

// ExecuteServingPark records uncertainty before invoking filesystem authority.
// The callback runs without a SQLite transaction: it reads the same journal
// under root's namespace and host locks. A pending replay revalidates authority
// and resumes the deterministic retirement; it never rolls filesystem work back.
func (s *JournalService) ExecuteServingPark(ctx context.Context, peer string, q JournalRequest, run func(context.Context) (ServingParkResult, error)) (JournalResponse, error) {
	out := JournalResponse{V: 1, ID: q.ID}
	if err := validateServingParkRequest(q); err != nil {
		return out, err
	}
	if s == nil || s.Launches == nil || s.Launches.db == nil || s.parkGate == nil || peer == "" || run == nil {
		return out, ErrUnsupported
	}
	select {
	case s.parkGate <- struct{}{}:
		defer func() { <-s.parkGate }()
	case <-ctx.Done():
		return out, ctx.Err()
	}
	pending, replayed, err := s.prepareServingPark(ctx, peer, q)
	if err != nil || replayed {
		return pending, err
	}
	result, err := run(ctx)
	if err != nil {
		// Even an identity refusal may follow a prior rename or a lost save
		// reply. Preserve the pending receipt and the exact request for replay.
		return out, errors.Join(ErrJournalUncertain, err)
	}
	out.ServingPark = &result
	body, err := json.Marshal(out)
	if err != nil {
		return out, errors.Join(ErrJournalUncertain, err)
	}
	changed, err := s.Launches.db.ExecContext(ctx, "UPDATE journal_requests SET response=? WHERE id=? AND peer=? AND digest=?", body, q.ID, peer, journalRequestDigest(q))
	if err != nil {
		return out, errors.Join(ErrJournalUncertain, err)
	}
	n, err := changed.RowsAffected()
	if err != nil || n != 1 {
		return out, ErrJournalUncertain
	}
	return out, nil
}

func (s *JournalService) prepareServingPark(ctx context.Context, peer string, q JournalRequest) (JournalResponse, bool, error) {
	out := JournalResponse{V: 1, ID: q.ID}
	tx, err := s.Launches.db.BeginTx(ctx, nil)
	if err != nil {
		return out, false, err
	}
	defer tx.Rollback()
	var oldPeer, digest string
	var body []byte
	err = tx.QueryRowContext(ctx, "SELECT peer,digest,response FROM journal_requests WHERE id=?", q.ID).Scan(&oldPeer, &digest, &body)
	found := err == nil
	if found {
		if oldPeer != peer || digest != journalRequestDigest(q) {
			return out, false, ErrChanged
		}
		if json.Unmarshal(body, &out) != nil || out.V != 1 || out.ID != q.ID {
			return out, false, ErrJournalUncertain
		}
		if out.Error == "" && out.ServingPark != nil {
			return out, true, nil
		}
		if out.Error != "uncertain" || out.ServingPark != nil {
			return out, false, ErrJournalUncertain
		}
	} else if !errors.Is(err, sql.ErrNoRows) {
		return out, false, err
	}
	now := time.Now()
	if s.Now != nil {
		now = s.Now()
	}
	var rejectBefore int64
	if err = tx.QueryRowContext(ctx, "SELECT reject_before FROM journal_receipt_retention WHERE id=1").Scan(&rejectBefore); err != nil {
		return out, false, err
	}
	age := now.Sub(time.Unix(0, q.IssuedAt))
	if q.IssuedAt <= rejectBefore || age > JournalRequestWindow || age < -JournalRequestWindow {
		return out, false, ErrJournalUncertain
	}
	if _, err = journalPrune(ctx, tx, rejectBefore, now); err != nil {
		return out, false, err
	}
	if !found {
		out.Error = "uncertain"
		body, err = json.Marshal(out)
		if err != nil {
			return out, false, err
		}
		if _, err = tx.ExecContext(ctx, "INSERT INTO journal_requests(id,peer,digest,response,created,issued) VALUES(?,?,?,?,?,?)", q.ID, peer, journalRequestDigest(q), body, now.UnixNano(), q.IssuedAt); err != nil {
			return out, false, err
		}
	}
	return out, false, tx.Commit()
}