package service

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/go-git/go-git/v5"
	"github.com/go-git/go-git/v5/plumbing"

	"sourcecraft.dev/bigbes/sr-ht-spec/core"
	"sourcecraft.dev/bigbes/sr-ht-spec/db"
	"sourcecraft.dev/bigbes/sr-ht-spec/gitx"
)

// Three systems are touched by a merge — git refs, the bleve index and
// Postgres — and none of it is transactional. The rule that makes it tractable:
//
//	Git refs are the source of truth for whether a proposal has merged. The
//	Postgres row is the source of truth for that a proposal exists and what it
//	is. The index and the render cache are pure caches.
//
// The reconciler is the backstop that repairs divergence, and it implements
// exactly four repairs — no more, because every additional guess about what a
// half-finished write meant is a way to invent state nobody wrote.

const (
	// DefaultReconcileInterval is how often the reconciler runs after startup.
	DefaultReconcileInterval = 15 * time.Minute

	// DefaultReconcileGrace is how long a proposal row with no branch is left
	// alone before it is deleted.
	//
	// The design's repair table has no grace period, and without one the
	// reconciler is actively destructive during ordinary operation: a proposal
	// is opened row-first, so every live propose passes through the exact state
	// ("open row, no branch") that the table says to delete. The window is
	// milliseconds wide and the reconciler runs on a timer, so it would be rare
	// — which makes it worse, not better, since it would destroy an agent's
	// work at random and never in a test.
	//
	// A row younger than this is therefore assumed to be in flight rather than
	// abandoned. It costs one extra reconcile cycle before a genuinely crashed
	// proposal is cleaned up, which nothing is waiting on.
	DefaultReconcileGrace = 5 * time.Minute
)

// RepairKind names one of the four repairs.
type RepairKind string

const (
	// RepairDeleteRow removes an `open` proposal row whose branch does not
	// exist: the daemon died between the row insert and the branch write. The
	// row holds no content, and the agent still holds the document it wanted to
	// write, so it re-proposes.
	RepairDeleteRow RepairKind = "delete-proposal-row"

	// RepairDeleteRef removes a proposals/* ref with no row. It is unreferenced
	// — the id is a Postgres serial, and title, rationale, base_rev, agent and
	// agent_session live nowhere in a ref — so its content is unrecoverable
	// anyway and recreating the row would mean inventing every field.
	RepairDeleteRef RepairKind = "delete-proposal-ref"

	// RepairMarkMerged transitions a row still `open` whose branch has merged
	// into the approved head. The ref is truth for merged-ness.
	RepairMarkMerged RepairKind = "mark-proposal-merged"

	// RepairReindex flags a space whose index stamp differs from its approved
	// head. Phase 2 owns the rebuild; the reconciler only reports the list.
	RepairReindex RepairKind = "reindex-space"
)

// Repair is one repair the reconciler decided on.
type Repair struct {
	Kind    RepairKind
	Space   core.SpaceRef
	SpaceID int

	// ProposalID is the proposal row's id, and zero for RepairReindex and for
	// an orphan ref whose name carries no parseable id.
	ProposalID int

	// Branch is the proposal branch this repair is about, empty for
	// RepairReindex.
	Branch string

	// Rev is the revision the repair records: the merge revision for
	// RepairMarkMerged, the approved head for RepairReindex.
	Rev string

	// Approval is the approval kind RepairMarkMerged records. See
	// PlanRepairs for why it is always core.ApprovalPolicy.
	Approval core.Approval

	// Reason is a human-readable sentence for the log.
	Reason string
}

func (r Repair) String() string {
	return fmt.Sprintf("%s %s: %s", r.Kind, r.Space, r.Reason)
}

// ProposalFact is what the reconciler observed about one proposal — its row,
// its branch, or both. Facts are gathered by I/O and consumed by PlanRepairs,
// which is pure so that the decision table can be exhaustively tested without a
// repository or a database.
type ProposalFact struct {
	// ID is the proposal row id, and the id parsed out of the branch name when
	// there is no row. Zero when the branch name carries no parseable id.
	ID int

	// Branch is the proposal branch name, "proposals/42".
	Branch string

	// HasRow and HasBranch record which halves exist. Both false is not a fact.
	HasRow    bool
	HasBranch bool

	// State is the row's state, meaningless when HasRow is false.
	State core.ProposalState

	// Created is when the row was inserted, for the grace window.
	Created time.Time

	// MergedIntoApproved reports whether the branch tip is reachable from the
	// approved head. Computed by I/O (it needs the object database) and passed
	// in, exactly as gitx.RefUpdate.FastForward is.
	MergedIntoApproved bool

	// BranchHead is the branch tip.
	BranchHead string

	// BaseRev is the proposal's recorded base, resolved to an object name, and
	// empty when it could not be resolved. See PlanRepairs for why a branch
	// still sitting on its base is not a merge.
	BaseRev string
}

// SpaceFacts is everything the reconciler observed about one space.
type SpaceFacts struct {
	Space   core.SpaceRef
	SpaceID int

	// ApprovedHead is the current tip of the approved branch.
	ApprovedHead string

	// IndexRev is the revision the global index currently reflects for this
	// space, empty when the space has never been indexed. Empty is stale by
	// construction, which is why a missing stamp is not treated as up to date.
	IndexRev string

	Proposals []ProposalFact

	// Now and Grace parameterize the grace window, so it is an input to the
	// decision rather than a clock read inside it.
	Now   time.Time
	Grace time.Duration
}

// PlanRepairs is the repair table, as a pure function.
//
// It implements exactly the four rows of the design's "Consistency and
// recovery" table and nothing else. States it does not name — a merged row
// whose branch is no longer an ancestor of the approved head, a rejected row
// whose branch still exists — are deliberately left alone: neither is a
// half-finished write, and repairing them would mean deciding something the
// design did not.
//
// "Merged" needs one qualification the design's table does not state, and
// without it the reconciler corrupts state during ordinary operation. A
// proposal branch is cut *at* the approved head, so between the cut and the
// agent's first commit its tip is trivially an ancestor of that head — and the
// literal rule "branch merged into the approved head, row still open" fires on
// a proposal that has not merged and has no content at all. The same holds
// forever after for a proposal whose agent never committed. A branch still
// sitting on its recorded base is therefore never treated as merged, and a base
// that could not be resolved is treated the same way: repairing on facts we
// could not establish is worse than leaving the row open for a human to see.
//
// RepairMarkMerged always records core.ApprovalPolicy. The ref proves the
// merge happened and nothing proves how it was authorized — the approval kind
// existed only in the memory of the process that died. Of the two available
// lies, "policy" is the safe one: recording "human" would launder unreviewed
// content as blessed, which is the exact failure the bimodal decision exists to
// prevent, while recording "policy" understates the review and puts the
// proposal in the policy-merged digest, where a human sees it again. Erring
// toward visibility is the whole point of the digest.
func PlanRepairs(f SpaceFacts) []Repair {
	var repairs []Repair
	base := Repair{Space: f.Space, SpaceID: f.SpaceID}

	for _, p := range f.Proposals {
		r := base
		r.ProposalID = p.ID
		r.Branch = p.Branch

		switch {
		case p.HasRow && p.HasBranch && p.State == core.StateOpen && p.MergedIntoApproved &&
			p.BaseRev != "" && p.BranchHead != p.BaseRev:
			r.Kind = RepairMarkMerged
			r.Rev = f.ApprovedHead
			r.Approval = core.ApprovalPolicy
			r.Reason = fmt.Sprintf("branch %s has merged into the approved head %s but the row is still open",
				p.Branch, short(f.ApprovedHead))
			repairs = append(repairs, r)

		case p.HasRow && !p.HasBranch && p.State == core.StateOpen:
			if f.Now.Sub(p.Created) < f.Grace {
				continue // in flight: the row is written before the branch
			}
			r.Kind = RepairDeleteRow
			r.Reason = fmt.Sprintf("row is open but branch %s does not exist; the agent re-proposes", p.Branch)
			repairs = append(repairs, r)

		case !p.HasRow && p.HasBranch:
			r.Kind = RepairDeleteRef
			r.Reason = fmt.Sprintf("branch %s has no row; its content is unrecoverable", p.Branch)
			repairs = append(repairs, r)
		}
	}

	if f.ApprovedHead != "" && f.IndexRev != f.ApprovedHead {
		r := base
		r.Kind = RepairReindex
		r.Rev = f.ApprovedHead
		r.Reason = fmt.Sprintf("index stamp %s differs from the approved head %s",
			stampOrNever(f.IndexRev), short(f.ApprovedHead))
		repairs = append(repairs, r)
	}
	return repairs
}

func short(rev string) string {
	if len(rev) > 8 {
		return rev[:8]
	}
	return rev
}

func stampOrNever(rev string) string {
	if rev == "" {
		return "(never indexed)"
	}
	return short(rev)
}

// ReconcileFailure is one thing the reconciler could not do. Failures never
// abort the run: a space with an unreadable repository must not stop the other
// spaces from being repaired.
type ReconcileFailure struct {
	Space  core.SpaceRef
	Repair *Repair // nil when the whole space could not be examined
	Err    error
}

func (f ReconcileFailure) Error() string {
	if f.Repair != nil {
		return fmt.Sprintf("%s: %v", f.Repair, f.Err)
	}
	return fmt.Sprintf("%s: %v", f.Space, f.Err)
}

// ReconcileReport is what one reconciler pass did.
type ReconcileReport struct {
	// Spaces is how many spaces were examined.
	Spaces int

	// Repaired lists the repairs that were applied.
	Repaired []Repair

	// Reindex lists spaces whose index is stale. They are reported, not
	// repaired: bleve is single-writer and Phase 2 owns the index. A caller
	// that has an indexer drives it from this list.
	Reindex []Repair

	// Failures lists what could not be examined or could not be repaired.
	Failures []ReconcileFailure
}

// Reconcile runs one pass: scan proposals/* refs and each space's approved
// head, compare against rows and index stamps, repair divergence.
//
// The read order is load-bearing. Refs are listed for every space *before* any
// proposal row is read, so a proposal opened concurrently can only ever look
// like "row with no branch" — which the grace window protects — and never like
// "branch with no row", which would delete a live agent's work. Reversing the
// two reads turns an ordinary concurrent propose into data loss.
func (s *Service) Reconcile(ctx context.Context) (*ReconcileReport, error) {
	spaces, err := s.ListSpaces(ctx)
	if err != nil {
		return nil, err
	}

	rep := &ReconcileReport{}

	// Pass one: every space's repository, approved head and proposal branches.
	type observed struct {
		space    *Space
		head     plumbing.Hash
		branches []gitx.Branch
	}
	seen := make([]observed, 0, len(spaces))
	for _, sp := range spaces {
		repo, err := s.openRepo(sp.Ref)
		if err != nil {
			rep.Failures = append(rep.Failures, ReconcileFailure{Space: sp.Ref, Err: err})
			continue
		}
		sp.Repo = repo
		head, err := repo.ApprovedHead(ctx)
		if err != nil {
			rep.Failures = append(rep.Failures, ReconcileFailure{Space: sp.Ref, Err: err})
			continue
		}
		branches, err := repo.ListProposalBranches(ctx)
		if err != nil {
			rep.Failures = append(rep.Failures, ReconcileFailure{Space: sp.Ref, Err: err})
			continue
		}
		seen = append(seen, observed{space: sp, head: head, branches: branches})
	}

	// Pass two: the rows, read strictly after every ref listing above.
	open, err := s.store.ListProposalsByState(ctx, core.StateOpen, 0)
	if err != nil {
		return nil, fmt.Errorf("service: list open proposals: %w", err)
	}
	openBySpace := make(map[int][]*db.Proposal, len(spaces))
	for _, p := range open {
		openBySpace[p.SpaceID] = append(openBySpace[p.SpaceID], p)
	}

	for _, o := range seen {
		rep.Spaces++
		facts, err := s.spaceFacts(ctx, o.space, o.head, o.branches, openBySpace[o.space.ID])
		if err != nil {
			rep.Failures = append(rep.Failures, ReconcileFailure{Space: o.space.Ref, Err: err})
			continue
		}
		for _, r := range PlanRepairs(facts) {
			if r.Kind == RepairReindex {
				rep.Reindex = append(rep.Reindex, r)
				continue
			}
			if err := s.applyRepair(ctx, o.space, r); err != nil {
				repair := r
				rep.Failures = append(rep.Failures, ReconcileFailure{
					Space: o.space.Ref, Repair: &repair, Err: err,
				})
				continue
			}
			rep.Repaired = append(rep.Repaired, r)
		}
	}
	return rep, nil
}

// spaceFacts turns one space's refs and rows into the facts PlanRepairs
// consumes, resolving the two things only I/O can answer: whether a branch has
// merged into the approved head, and whether a branch without an *open* row has
// any row at all.
func (s *Service) spaceFacts(ctx context.Context, sp *Space, head plumbing.Hash,
	branches []gitx.Branch, openRows []*db.Proposal) (SpaceFacts, error) {

	facts := SpaceFacts{
		Space:        sp.Ref,
		SpaceID:      sp.ID,
		ApprovedHead: head.String(),
		Now:          s.now(),
		Grace:        s.grace,
	}

	stamp, err := s.store.GetIndexStamp(ctx, sp.ID)
	switch {
	case err == nil:
		facts.IndexRev = stamp.Rev
	case errors.Is(err, db.ErrNotFound):
		// Never indexed. Left empty, which PlanRepairs reads as stale.
	default:
		return SpaceFacts{}, fmt.Errorf("service: read index stamp for %s: %w", sp.Ref, err)
	}

	byBranch := make(map[string]gitx.Branch, len(branches))
	for _, b := range branches {
		byBranch[b.Name] = b
	}

	rowBranches := make(map[string]bool, len(openRows))
	for _, row := range openRows {
		rowBranches[row.Branch] = true
		fact := ProposalFact{
			ID:      row.ID,
			Branch:  row.Branch,
			HasRow:  true,
			State:   row.State,
			Created: row.Created,
		}
		if b, ok := byBranch[row.Branch]; ok {
			fact.HasBranch = true
			fact.BranchHead = b.Head.String()
			merged, err := sp.Repo.IsAncestor(ctx, b.Head, head)
			if err != nil {
				return SpaceFacts{}, fmt.Errorf("service: ancestry of %s in %s: %w", row.Branch, sp.Ref, err)
			}
			fact.MergedIntoApproved = merged
			// Resolved rather than compared as a string: base_rev is whatever
			// the agent sent as If-Match, and an abbreviated spelling of the
			// branch tip would otherwise read as "the branch has commits".
			//
			// A base that is no longer in the repository leaves this empty,
			// which PlanRepairs reads as "do not repair" — the row stays open
			// where a human can see it. Any other failure is a real read error
			// and is surfaced rather than silently disarming the check.
			base, err := sp.Repo.ResolveRev(ctx, row.BaseRev)
			switch {
			case err == nil:
				fact.BaseRev = base.String()
			case errors.Is(err, gitx.ErrNotFound), errors.Is(err, gitx.ErrBadRev):
			default:
				return SpaceFacts{}, fmt.Errorf("service: resolve base %q of %s in %s: %w",
					row.BaseRev, row.Branch, sp.Ref, err)
			}
		}
		facts.Proposals = append(facts.Proposals, fact)
	}

	for _, b := range branches {
		if rowBranches[b.Name] {
			continue
		}
		// No *open* row claims this branch. A merged or rejected proposal keeps
		// its branch and its row, so before calling the ref an orphan we ask
		// whether any row owns it. Getting this wrong deletes the branch of an
		// already-merged proposal.
		id, ok := gitx.ParseProposalBranch(b.Name)
		if ok {
			row, err := s.store.GetProposal(ctx, int(id))
			switch {
			case err == nil && row.SpaceID == sp.ID && row.Branch == b.Name:
				facts.Proposals = append(facts.Proposals, ProposalFact{
					ID: row.ID, Branch: b.Name, HasRow: true, HasBranch: true,
					State: row.State, Created: row.Created, BranchHead: b.Head.String(),
				})
				continue
			case err == nil, errors.Is(err, db.ErrNotFound):
				// Either no row at all, or a row that belongs to another space
				// or another branch — both mean this ref is unreferenced.
			default:
				return SpaceFacts{}, fmt.Errorf("service: look up proposal %d for %s: %w", id, sp.Ref, err)
			}
		}
		facts.Proposals = append(facts.Proposals, ProposalFact{
			ID: int(idOrZero(b.Name)), Branch: b.Name, HasBranch: true, BranchHead: b.Head.String(),
		})
	}
	return facts, nil
}

func idOrZero(branch string) int64 {
	id, ok := gitx.ParseProposalBranch(branch)
	if !ok {
		return 0
	}
	return id
}

// applyRepair executes one repair. RepairReindex never reaches it — Phase 2
// owns the index — and an unknown kind is an error rather than a no-op, so a
// repair added to PlanRepairs without an implementation fails loudly.
func (s *Service) applyRepair(ctx context.Context, sp *Space, r Repair) error {
	switch r.Kind {
	case RepairMarkMerged:
		return s.store.MarkProposalMerged(ctx, r.ProposalID, r.Approval, r.Rev)
	case RepairDeleteRow:
		return s.deleteOpenProposalRow(ctx, r.ProposalID)
	case RepairDeleteRef:
		return s.deleteProposalRef(ctx, sp, r.Branch)
	default:
		return fmt.Errorf("service: no implementation for repair %q", r.Kind)
	}
}

// deleteOpenProposalRow removes a proposal row that never got a branch.
//
// The SQL is written here rather than called on db.Store because db/ exposes no
// delete — the corrected repair table needs one and the persistence layer was
// built before the table was corrected. It belongs in db/ as DeleteOpenProposal
// and should move there; it lives in this file so that the repair the design
// specifies actually happens rather than silently not happening.
//
// The `state = 'open'` guard is in the statement, not in Go: a check-then-write
// would let a merge land in between and delete the row of a proposal that had
// just succeeded.
func (s *Service) deleteOpenProposalRow(ctx context.Context, id int) error {
	const q = `DELETE FROM proposal WHERE id = $1 AND state = 'open'`
	res, err := s.q.ExecContext(ctx, q, id)
	if err != nil {
		return fmt.Errorf("service: delete proposal %d: %w", id, err)
	}
	n, err := res.RowsAffected()
	if err != nil {
		return fmt.Errorf("service: delete proposal %d: rows affected: %w", id, err)
	}
	if n == 0 {
		return fmt.Errorf("service: proposal %d is no longer open; not deleted", id)
	}
	return nil
}

// deleteProposalRef removes an unreferenced proposals/* branch.
//
// Like the row delete above, this primitive is missing from the layer that
// should own it: gitx exposes branch creation and the merge but no delete. It
// is written here against go-git directly, under gitx's own per-space write
// lock, so it still excludes the merge path and a concurrent in-process write —
// but it should move into gitx, where the compare-and-swap retry logic for the
// two-writer problem already lives.
//
// The namespace check is not defence in depth, it is the only thing standing
// between a caller bug and a deleted approved branch.
func (s *Service) deleteProposalRef(ctx context.Context, sp *Space, branch string) error {
	if !gitx.IsProposalBranch(branch) {
		return fmt.Errorf("service: refusing to delete %q: only %s* branches are deletable",
			branch, gitx.ProposalPrefix)
	}
	return sp.Repo.WithLock(ctx, func(ctx context.Context) error {
		repo, err := git.PlainOpen(sp.Repo.Dir())
		if err != nil {
			return fmt.Errorf("service: open %s: %w", sp.Repo.Dir(), err)
		}
		name := plumbing.NewBranchReferenceName(branch)
		if err := repo.Storer.RemoveReference(name); err != nil {
			return fmt.Errorf("service: delete %s in %s: %w", name, sp.Ref, err)
		}
		return nil
	})
}

// RunReconciler runs the reconciler at startup and then on a ticker, until ctx
// is cancelled. report is called with the outcome of every pass; a nil report
// callback discards it.
//
// Running at startup is the half that matters: a daemon killed mid-merge
// repairs itself on the next boot with no manual intervention. The ticker
// catches the rest — a crash that leaves the daemon running, or a repair that
// failed once and succeeds later.
func (s *Service) RunReconciler(ctx context.Context, interval time.Duration, report func(*ReconcileReport, error)) {
	if interval <= 0 {
		interval = DefaultReconcileInterval
	}
	run := func() {
		rep, err := s.Reconcile(ctx)
		if report != nil {
			report(rep, err)
		}
	}
	run()

	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			run()
		}
	}
}