// Package dbverify is the fleet's restore-drill record: which databases have
// been PROVEN restorable, when, and which could not be decided.
//
// It exists as its own package for the reason internal/usage does — the
// writer (the box-side `carlos ops restore-verify` pass) and any future
// reader (the console) must agree on one shape and one key without either
// importing the other's machinery.
//
// The vocabulary is deliberately three-valued, following
// usage.AppStorage.StoreUnswept: a replica is `ok`, `failed`, or
// `unverified` — never a silent absence and never a 0/false that reads as
// "fine". A pass that could not reach S3 has learned NOTHING about the
// replica, and saying so is the whole point of keeping this record.
package dbverify

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"time"

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

// Key is where the record lives in the deployment bucket, beside
// control/serial and control/registry-snapshot.json.
const Key = "control/restore-verify.json"

// Verdict is one database's standing.
type Verdict struct {
	// OK is true only when the replica was actually restored and passed its
	// integrity check on this box, at VerifiedAt.
	OK bool `json:"ok"`
	// Unverified marks "we could not tell" — an S3 error, a prefix that has
	// never been replicated to, or a host the run's cap did not reach. It is
	// never a failure and never a pass.
	Unverified bool `json:"unverified,omitempty"`
	// VerifiedAt is when this verdict was reached — NOT when the replica was
	// written. An old timestamp on an ok verdict is itself a finding.
	VerifiedAt time.Time `json:"verified_at"`
	// Detail carries the failure or the reason it could not be decided.
	Detail string `json:"detail,omitempty"`
}

// Record is the whole box's standing at CheckedAt.
type Record struct {
	V         int                `json:"v"` // 1
	Box       string             `json:"box"`
	CheckedAt time.Time          `json:"checked_at"`
	Hosts     map[string]Verdict `json:"hosts"`
}

// Merge folds this run's verdicts into the prior record. prior may be nil.
//
// Two rules, both about not overstating coverage:
//
//   - A capped run visits only some hosts, so untouched hosts KEEP their
//     previous verdict — with its old timestamp, which is what makes staleness
//     visible rather than hidden behind a fresh CheckedAt.
//   - A host no longer in `fleet` is DROPPED. A retired instance's last "ok"
//     would otherwise sit in the record forever, padding the coverage of a
//     database that no longer exists.
func Merge(prior *Record, box string, now time.Time, fleet []string, results map[string]Verdict) *Record {
	rec := &Record{V: 1, Box: box, CheckedAt: now.UTC(), Hosts: map[string]Verdict{}}
	inFleet := map[string]bool{}
	for _, h := range fleet {
		inFleet[h] = true
	}
	if prior != nil {
		for host, v := range prior.Hosts {
			if inFleet[host] {
				rec.Hosts[host] = v
			}
		}
	}
	for host, v := range results {
		if !inFleet[host] {
			continue
		}
		rec.Hosts[host] = v
	}
	return rec
}

// Failures counts the verdicts that are an alarm — restored and damaged.
// Unverified rows are deliberately not counted here: they are a coverage
// problem, not a corruption one, and conflating them would either cry wolf or
// train operators to ignore a red run.
func (r *Record) Failures() int {
	n := 0
	for _, v := range r.Hosts {
		if !v.OK && !v.Unverified {
			n++
		}
	}
	return n
}

// Unverified counts the verdicts the pass could not decide.
func (r *Record) Unverified() int {
	n := 0
	for _, v := range r.Hosts {
		if v.Unverified {
			n++
		}
	}
	return n
}

// Load reads the current record. A missing key is (nil, nil) — a deployment
// that has never run the drill, not an error.
func Load(ctx context.Context, s release.Store) (*Record, error) {
	body, _, err := s.Get(ctx, Key)
	if errors.Is(err, os.ErrNotExist) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	var rec Record
	if err := json.Unmarshal(body, &rec); err != nil {
		return nil, fmt.Errorf("parse %s: %w", Key, err)
	}
	return &rec, nil
}

// RemoveHost drops host's verdict from the record: a moved instance's
// replica is no longer this fleet's to verify where it was (moves spec
// §6.3, §7.6 local). The load, the delete and the put run inside each CAS
// attempt (moves Plan 2 ruling G31), so a record another writer changed
// between the read and the write is re-read, never overwritten with a stale
// body. No record, or no entry for the host, is nothing to do.
func RemoveHost(ctx context.Context, s release.Store, host string) error {
	for attempt := 0; attempt < 3; attempt++ {
		body, etag, err := s.Get(ctx, Key)
		if errors.Is(err, os.ErrNotExist) {
			return nil
		}
		if err != nil {
			return fmt.Errorf("get %s: %w", Key, err)
		}
		var rec Record
		if err := json.Unmarshal(body, &rec); err != nil {
			return fmt.Errorf("parse %s: %w", Key, err)
		}
		if _, ok := rec.Hosts[host]; !ok {
			return nil
		}
		delete(rec.Hosts, host)
		next, err := json.Marshal(&rec)
		if err != nil {
			return err
		}
		if _, err := s.Put(ctx, Key, next, etag); err == nil {
			return nil
		} else if !errors.Is(err, release.ErrCASConflict) {
			return fmt.Errorf("put %s: %w", Key, err)
		}
	}
	return fmt.Errorf("put %s: lost three CAS races", Key)
}

// Publish writes rec, replacing whatever is there. Three attempts on a lost
// CAS race, the recipe usage.UpsertEdgeDay and host.putReplacing both use.
func Publish(ctx context.Context, s release.Store, rec *Record) error {
	body, err := json.Marshal(rec)
	if err != nil {
		return err
	}
	for attempt := 0; attempt < 3; attempt++ {
		_, etag, err := s.Get(ctx, Key)
		switch {
		case errors.Is(err, os.ErrNotExist):
			etag = ""
		case err != nil:
			return fmt.Errorf("get %s: %w", Key, err)
		}
		if _, err := s.Put(ctx, Key, body, etag); err == nil {
			return nil
		} else if !errors.Is(err, release.ErrCASConflict) {
			return fmt.Errorf("put %s: %w", Key, err)
		}
	}
	return fmt.Errorf("put %s: lost three CAS races", Key)
}