package store

import (
	"database/sql"
	"errors"
)

// chunkstatus.go: the per-chunk×stage disposition record (Веха 2, D2). It is a
// materialized RESOLVE over the append-only checkpoints (keyed by request_hash,
// which already carries `attempt`), NOT a column on checkpoints. Because it is
// derivable from the checkpoints, losing it to a kill -9 self-heals: on resume
// the runner re-classifies the checkpoint text (no re-billing) and re-upserts
// the row. `disposition` is the axis the runner loop reads BEFORE rendering, so
// a terminally-flagged chunk is never re-attacked into an infinite paid loop.
//
// The store is dumb storage here: disposition/flag_reason are plain strings; the
// typed FlagReason/Disposition constants and the classifier live in the pipeline
// package.

// ChunkStatus is one (book, chapter, chunk, stage) disposition row.
type ChunkStatus struct {
	BookID          string
	Chapter         int
	ChunkIdx        int
	Stage           string
	SnapshotID      string
	ContentHash     string  // signature of the rendered msgs (the source is NOT in the snapshot); guards the resume fast-path against serving a stale translation after a source edit
	Disposition     string  // ok | flagged | skipped
	FlagReason      string  // "" when ok
	Attempts        int     // number of attempts made for this chunk×stage
	FinalHash       string  // request_hash of the authoritative checkpoint (ok path)
	CostUSD         float64 // sum across all attempts (F3-honest)
	Detail          string
	Escalated       bool   // a single-hop fallback was used for this chunk×stage (D12/D15.3 telemetry)
	EscalationModel string // the fallback model that answered ("" when not escalated)
}

// UpsertChunkStatus writes (or overwrites) the disposition row. Overwrite is the
// resolve semantics: re-running classify over the same checkpoints reproduces the
// same row, and a --resnapshot re-processing legitimately replaces a stale row.
func (s *Store) UpsertChunkStatus(cs ChunkStatus) error {
	ctx, cancel := opContext()
	defer cancel()
	_, err := s.w.ExecContext(ctx, `
		INSERT INTO chunk_status (
			book_id, chapter, chunk_idx, stage, snapshot_id, content_hash,
			disposition, flag_reason, attempts, final_hash, cost_usd, detail,
			escalated, escalation_model, updated_at
		) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now'))
		ON CONFLICT (book_id, chapter, chunk_idx, stage) DO UPDATE SET
			snapshot_id      = excluded.snapshot_id,
			content_hash     = excluded.content_hash,
			disposition      = excluded.disposition,
			flag_reason      = excluded.flag_reason,
			attempts         = excluded.attempts,
			final_hash       = excluded.final_hash,
			cost_usd         = excluded.cost_usd,
			detail           = excluded.detail,
			escalated        = excluded.escalated,
			escalation_model = excluded.escalation_model,
			updated_at       = excluded.updated_at`,
		cs.BookID, cs.Chapter, cs.ChunkIdx, cs.Stage, cs.SnapshotID, cs.ContentHash,
		cs.Disposition, cs.FlagReason, cs.Attempts, cs.FinalHash, cs.CostUSD, cs.Detail,
		boolToInt(cs.Escalated), cs.EscalationModel)
	return err
}

// GetChunkStatus returns the disposition row for a chunk×stage, or nil if none.
func (s *Store) GetChunkStatus(bookID string, chapter, chunkIdx int, stage string) (*ChunkStatus, error) {
	ctx, cancel := opContext()
	defer cancel()
	cs := ChunkStatus{BookID: bookID, Chapter: chapter, ChunkIdx: chunkIdx, Stage: stage}
	var escalated int
	err := s.r.QueryRowContext(ctx, `
		SELECT snapshot_id, content_hash, disposition, flag_reason, attempts, final_hash, cost_usd, detail,
		       escalated, escalation_model
		FROM chunk_status
		WHERE book_id = ? AND chapter = ? AND chunk_idx = ? AND stage = ?`,
		bookID, chapter, chunkIdx, stage).Scan(
		&cs.SnapshotID, &cs.ContentHash, &cs.Disposition, &cs.FlagReason, &cs.Attempts, &cs.FinalHash, &cs.CostUSD, &cs.Detail,
		&escalated, &cs.EscalationModel)
	cs.Escalated = escalated != 0
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return &cs, nil
}

// ChunkStatusesForBook returns every disposition row for a book, ordered for a
// stable report (tmctl report: the flag section).
func (s *Store) ChunkStatusesForBook(bookID string) ([]ChunkStatus, error) {
	ctx, cancel := opContext()
	defer cancel()
	rows, err := s.r.QueryContext(ctx, `
		SELECT chapter, chunk_idx, stage, snapshot_id, content_hash, disposition, flag_reason,
		       attempts, final_hash, cost_usd, detail, escalated, escalation_model
		FROM chunk_status WHERE book_id = ?
		ORDER BY chapter, chunk_idx, stage`, bookID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []ChunkStatus
	for rows.Next() {
		cs := ChunkStatus{BookID: bookID}
		var escalated int
		if err := rows.Scan(&cs.Chapter, &cs.ChunkIdx, &cs.Stage, &cs.SnapshotID, &cs.ContentHash,
			&cs.Disposition, &cs.FlagReason, &cs.Attempts, &cs.FinalHash, &cs.CostUSD, &cs.Detail,
			&escalated, &cs.EscalationModel); err != nil {
			return nil, err
		}
		cs.Escalated = escalated != 0
		out = append(out, cs)
	}
	return out, rows.Err()
}

// ResetChunkStages deletes the chunk_status rows AND the checkpoints for the given stages of
// ONE chunk, in a single transaction — the durable half of `tmctl redrive` (D15.3). After
// this, the next translate re-derives those stages: their chunk_status is gone (the runner
// re-runs instead of resuming the flag) and their checkpoints are gone (a FRESH provider call
// is made, not a deterministic replay of the flagged completion). Only the passed stages are
// touched — an upstream OK stage keeps its checkpoint and resumes at $0 (D12: never re-pay
// DispOK work). MONEY (documented, honest): the money already spent on the discarded attempts
// STAYS committed in `spend` (it was really billed at the provider), so the spend ceilings
// remain honest; only the resume cache and the escalation-budget accounting are reset, which
// is the "fresh retry/escalation budget" redrive grants an explicit operator command. Thus
// after a redrive committed(spend) >= SUM(checkpoints) — the safe direction (a ceiling never
// under-counts). Checkpoints are joined to jobs by (book, chapter, stage); request_log
// (append-only telemetry) is deliberately NOT touched, so the paid history is still auditable.
func (s *Store) ResetChunkStages(bookID string, chapter, chunkIdx int, stages []string) error {
	if len(stages) == 0 {
		return nil
	}
	ctx, cancel := opContext()
	defer cancel()
	tx, err := s.w.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Rollback()
	for _, stage := range stages {
		if _, err := tx.ExecContext(ctx, `
			DELETE FROM checkpoints
			WHERE chunk_idx = ? AND job_id IN (
				SELECT id FROM jobs WHERE book_id = ? AND chapter = ? AND stage = ?)`,
			chunkIdx, bookID, chapter, stage); err != nil {
			return err
		}
		if _, err := tx.ExecContext(ctx, `
			DELETE FROM chunk_status
			WHERE book_id = ? AND chapter = ? AND chunk_idx = ? AND stage = ?`,
			bookID, chapter, chunkIdx, stage); err != nil {
			return err
		}
	}
	return tx.Commit()
}