// replayreverse runs the §10 reverse-replay controlled experiment on a
// single pricing_comparison_logs row: it reconstructs the old path's
// state S0 from the row's own persisted snapshot, runs the new path's
// selection and pricing rules over S0, and compares the output against
// the old path's recorded row with the production comparator.
//
// Input on stdin (or first file argument) is one JSON object in the
// exact shape of a model/billing/testdata/replay_backtest.jsonl line:
//
//	mysql ... -e "select pricing, raw, diffs from pricing_comparison_logs where id=123" \
//	  | jq -c '.[0]' | go run ./cmd/replayreverse
//
// Concurrent window rows (optional, --window) are raw_user_unit_costs
// rows of the same tenant priced inside (t0, t1] — one JSON object per
// line, exactly the shape of a raw row. Without them the experiment
// runs under the quiet-window approximation and busy windows degrade
// to inconclusive via the selection guard:
//
//	mysql ... -e "select * from user_unit_costs where tenant_id=T and \
//	  created_at between from and to" | jq -c '.' > window.jsonl
//	go run ./cmd/replayreverse --window window.jsonl < row.json
//
// Verdicts: equivalent (logics agree on S0) / divergent (logic-fork
// candidate — triage by hand) / inconclusive (S0 not recoverable).
package main

import (
	"bufio"
	"encoding/json"
	"fmt"
	"io"
	"os"

	"github.com/pingcap/dbaas/billing2/db/entity"
	"github.com/pingcap/dbaas/billing2/model/billing"
	"github.com/pingcap/dbaas/billing2/model/billing/pipeline"
)

func main() {
	var in io.Reader = os.Stdin
	var windowPath string
	args := os.Args[1:]
	for len(args) > 0 {
		switch args[0] {
		case "--window":
			if len(args) < 2 {
				fatal("--window needs a file path")
			}
			windowPath = args[1]
			args = args[2:]
		default:
			f, err := os.Open(args[0])
			if err != nil {
				fatal("open input: %v", err)
			}
			defer f.Close()
			in = f
			args = args[1:]
		}
	}

	rawJSON, err := io.ReadAll(in)
	if err != nil {
		fatal("read input: %v", err)
	}
	var row struct {
		ID      int64  `json:"id"`
		Pricing string `json:"pricing"`
		Raw     string `json:"raw"`
		Diffs   string `json:"diffs"`
	}
	if err := json.Unmarshal(rawJSON, &row); err != nil {
		fatal("parse input (expected {pricing, raw, diffs} JSON): %v", err)
	}

	var pe entity.PricingEvent
	if err := json.Unmarshal([]byte(row.Pricing), &pe); err != nil {
		fatal("parse pricing side: %v", err)
	}
	var raw entity.RawUserUnitCost
	if err := json.Unmarshal([]byte(row.Raw), &raw); err != nil {
		fatal("parse raw side: %v", err)
	}
	entries, err := pipeline.ParsePricingDetail(pe.PricingDetail)
	if err != nil {
		fatal("parse pricing detail: %v", err)
	}

	var window []entity.RawUserUnitCost
	if windowPath != "" {
		window, err = loadWindow(windowPath)
		if err != nil {
			fatal("load window: %v", err)
		}
	}

	verdict, trace, diffs := billing.ReverseReplayArtifact(&pe, &raw, entries, window)
	switch verdict {
	case billing.ReverseVerdictEquivalent:
		fmt.Println("verdict: equivalent — new-path logic on S0 reproduces the old path's output")
	case billing.ReverseVerdictDivergent:
		fmt.Println("verdict: divergent — logic-fork candidate, triage by hand:")
		for _, d := range diffs {
			fmt.Printf("  diff %s: pipeline(replayed)=%s raw=%s\n", d.Field, d.Pipeline, d.Raw)
		}
	default:
		fmt.Println("verdict: inconclusive — S0 not recoverable from persisted data")
	}
	fmt.Printf("trace (cost_id=%s tenant=%d id=%d window=%d):\n", pe.CostID, pe.TenantID, row.ID, len(window))
	for _, line := range billing.ReplayTraceDetails(trace) {
		fmt.Printf("  %s\n", line)
	}
}

func loadWindow(path string) ([]entity.RawUserUnitCost, error) {
	f, err := os.Open(path)
	if err != nil {
		return nil, err
	}
	defer f.Close()
	var out []entity.RawUserUnitCost
	scanner := bufio.NewScanner(f)
	scanner.Buffer(make([]byte, 0, 1024*1024), 16*1024*1024)
	for scanner.Scan() {
		line := scanner.Bytes()
		if len(line) == 0 {
			continue
		}
		var w entity.RawUserUnitCost
		if err := json.Unmarshal(line, &w); err != nil {
			return nil, fmt.Errorf("window line not a raw row: %w", err)
		}
		out = append(out, w)
	}
	return out, scanner.Err()
}

func fatal(format string, args ...any) {
	fmt.Fprintf(os.Stderr, "replayreverse: "+format+"\n", args...)
	os.Exit(1)
}