package console

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"log"
	"sort"
	"strings"
	"time"

	"github.com/carlosframework/platform/internal/alerting"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostops"
)

const (
	claimLossPageInterval      = time.Hour
	claimLossPageDailyInterval = 24 * time.Hour
	claimLossPageBudget        = 30 * time.Second
)

// claimLossReadBudget bounds one record's read inside a scan. A record whose
// read keeps stalling then fails on its own and the scan moves past it,
// retrying it next lap, instead of spending the whole scan budget on it every
// time and starving the records behind it.
var claimLossReadBudget = 5 * time.Second

// Recovery schedules one bounded scan. A slow sink holds only this worker;
// later recovery passes skip scheduling until it exits.
func (s *Server) scheduleClaimLossPages(ctx context.Context) {
	if !s.cfg.HostClaim || ctx.Err() != nil {
		return
	}
	s.claimLossPageMu.Lock()
	defer s.claimLossPageMu.Unlock()
	if s.claimLossPageDone != nil {
		select {
		case <-s.claimLossPageDone:
		default:
			return
		}
	}
	done := make(chan struct{})
	s.claimLossPageDone = done
	go func() {
		defer close(done)
		pass, cancel := context.WithTimeout(ctx, claimLossPageBudget)
		defer cancel()
		if err := s.pageClaimLosses(pass); err != nil {
			log.Printf("console: claim recovery: %v", err)
		}
	}()
}

func (s *Server) pageClaimLosses(ctx context.Context) error {
	return s.pageClaimLossesAt(ctx, time.Now)
}

// Scan records independently of claims: loss evidence outlives a holder or
// operation, and may still need attention after its claim has gone.
func (s *Server) pageClaimLossesAt(ctx context.Context, now func() time.Time) error {
	if !s.cfg.HostClaim {
		return nil
	}
	if s.cfg.Releases == nil || s.store == nil || len(s.cfg.ReleasePub) == 0 {
		return hostops.ErrContract
	}
	var after string
	if err := s.store.db.QueryRowContext(ctx, `SELECT last_key FROM claim_loss_pages WHERE rule = 'scan'`).Scan(&after); err != nil && !errors.Is(err, sql.ErrNoRows) {
		return err
	}
	keys, err := s.cfg.Releases.List(ctx, "control/host-loss/")
	if err != nil {
		return err
	}
	// Store listings need not be ordered. Visit each key once, wrapping even
	// when the saved key has since been removed or earlier keys were added.
	sort.Strings(keys)
	start := sort.Search(len(keys), func(i int) bool { return keys[i] > after })
	var errs []error
	for i := range keys {
		if err := ctx.Err(); err != nil {
			return errors.Join(append(errs, err)...)
		}
		key := keys[(start+i)%len(keys)]
		errs = append(errs, s.pageClaimLossRecord(ctx, key, now))
		if err := ctx.Err(); err != nil {
			return errors.Join(append(errs, err)...)
		}
		// Never advance past canceled work. Other record errors are retried on
		// the next lap so one bad record cannot block everything after it.
		if _, err := s.store.db.ExecContext(ctx, `INSERT INTO claim_loss_pages (rule, first_attempted_at, attempted_at, last_key) VALUES ('scan', 0, 0, ?)
ON CONFLICT(rule) DO UPDATE SET last_key = excluded.last_key`, key); err != nil {
			return errors.Join(append(errs, err)...)
		}
	}
	return errors.Join(errs...)
}

func (s *Server) pageClaimLossRecord(ctx context.Context, key string, now func() time.Time) error {
	id := strings.TrimSuffix(strings.TrimPrefix(key, "control/host-loss/"), ".json")
	exact, err := hostclaim.LossRecordKey(id)
	if err != nil || exact != key {
		return hostclaim.ErrLossRecord
	}
	read, cancel := context.WithTimeout(ctx, claimLossReadBudget)
	observed, err := hostclaim.ReadLossRecord(read, s.cfg.Releases, s.cfg.ReleasePub, id)
	cancel()
	if errors.Is(err, hostclaim.ErrLossRecordAbsent) {
		return nil
	}
	if err != nil {
		return err
	}
	r := observed.Record
	var errs []error
	if r.AlertsPending {
		subject, body := claimLossPendingPage(r.Host, r.ID)
		errs = append(errs, s.pageClaimLoss(ctx, "pending", r, now, subject, body))
	}
	if r.Timers.Knowledge == hostclaim.TimersPartial || r.Timers.Knowledge == hostclaim.TimersUnknown {
		subject, body := claimTimerLossPage(r.Host, r.ID)
		errs = append(errs, s.pageClaimLoss(ctx, "timers", r, now, subject, body))
	}
	return errors.Join(errs...)
}

func (s *Server) pageClaimLoss(ctx context.Context, condition string, r hostclaim.LossRecord, now func() time.Time, subject, body string) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	attempted, err := s.store.ClaimLossPageAttempt(ctx, condition+" "+r.ID, now())
	if err != nil || !attempted {
		return err
	}
	return alerting.Page(ctx, alerting.LevelWarning, subject, body)
}

// ClaimLossPageAttempt reserves the attempt before delivery, including failed
// sends. The conditional write also serializes overlapping recovery passes.
func (s *Store) ClaimLossPageAttempt(ctx context.Context, rule string, now time.Time) (bool, error) {
	result, err := s.db.ExecContext(ctx, `INSERT INTO claim_loss_pages (rule, first_attempted_at, attempted_at) VALUES (?, ?, ?)
ON CONFLICT(rule) DO UPDATE SET attempted_at = excluded.attempted_at
WHERE claim_loss_pages.attempted_at <= CASE
WHEN claim_loss_pages.first_attempted_at > ? THEN ? ELSE ? END`,
		rule, now.UnixNano(), now.UnixNano(), now.Add(-claimLossPageDailyInterval).UnixNano(),
		now.Add(-claimLossPageInterval).UnixNano(), now.Add(-claimLossPageDailyInterval).UnixNano())
	if err != nil {
		return false, err
	}
	n, err := result.RowsAffected()
	return n == 1, err
}

// Draft operator copy. The controller collects these strings for approval.
func claimLossPendingPage(host, id string) (subject, body string) {
	return fmt.Sprintf("Loss recorded for %s", host),
		fmt.Sprintf("A loss was recorded for host %s. Review loss record %s and check what needs recovery.", host, id)
}

func claimTimerLossPage(host, id string) (subject, body string) {
	return fmt.Sprintf("Timers may be lost for %s", host),
		fmt.Sprintf("Timers may have been lost with the box serving host %s. Check loss record %s. Check which timers are missing and re-arm them.", host, id)
}