package registry

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
)

var ErrClaimOwnership = errors.New("registry: claim row ownership changed or unknown")

type ClaimOwnership struct {
	ClaimIncarnation string
	ClaimStamp       uint64
	ClaimOwner       string
	ClaimRowID       string
}

func (o ClaimOwnership) Valid() bool {
	return o.ClaimIncarnation != "" && o.ClaimStamp != 0 && o.ClaimOwner != "" && o.ClaimRowID != ""
}

type ClaimRow struct {
	AllowAbsent  bool `json:",omitempty"`
	Kind         string
	Host         string
	Name         string
	Registration string
	Ownership    ClaimOwnership
}
type ClaimRowArchive struct {
	Row      ClaimRow
	Contents json.RawMessage
}

func (r *Registry) migrateClaimOwnership() error {
	for _, table := range []string{"routes", "timers"} {
		for _, column := range []string{"claim_incarnation TEXT NOT NULL DEFAULT ''", "claim_stamp INTEGER NOT NULL DEFAULT 0", "claim_owner TEXT NOT NULL DEFAULT ''", "claim_row_id TEXT NOT NULL DEFAULT ''"} {
			if _, err := r.sql.Exec("ALTER TABLE " + table + " ADD COLUMN " + column); err != nil && !strings.Contains(err.Error(), "duplicate column") {
				return err
			}
		}
	}
	_, err := r.sql.Exec(`CREATE TABLE IF NOT EXISTS claim_row_retirements (
 journal TEXT NOT NULL, kind TEXT NOT NULL, host TEXT NOT NULL, name TEXT NOT NULL,
 row_id TEXT NOT NULL, archive BLOB NOT NULL,
 PRIMARY KEY(journal,kind,host,name,row_id))`)
	return err
}

// AdoptClaimRoute is called only by the complete-census backfill coordinator.
// The full observed row is compared inside the write transaction.
func (r *Registry) AdoptClaimRoute(ctx context.Context, observed Route, owner ClaimOwnership) error {
	if r.remote != nil {
		_, err := r.remote.call(ctx, remoteRequest{Method: "AdoptClaimRoute", Observed: observed, Owner: owner}, nil)
		return err
	}

	if !owner.Valid() || observed.ClaimOwnership != (ClaimOwnership{}) {
		return ErrClaimOwnership
	}
	tx, err := r.beginClaimTransaction(ctx, true)
	if err != nil {
		return err
	}
	defer tx.Rollback()
	current, err := scanRoute(tx.QueryRowContext(ctx, `SELECT `+routeCols+` FROM routes WHERE host=?`, observed.Host).Scan)
	if err != nil {
		return err
	}
	a, _ := json.Marshal(current)
	b, _ := json.Marshal(observed)
	if !bytes.Equal(a, b) {
		return ErrClaimOwnership
	}
	res, err := tx.ExecContext(ctx, `UPDATE routes SET claim_incarnation=?,claim_stamp=?,claim_owner=?,claim_row_id=? WHERE host=? AND claim_owner='' AND claim_stamp=0 AND claim_incarnation='' AND claim_row_id=''`, owner.ClaimIncarnation, owner.ClaimStamp, owner.ClaimOwner, owner.ClaimRowID, observed.Host)
	if err != nil {
		return err
	}
	n, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if n != 1 {
		return ErrClaimOwnership
	}
	return tx.Commit()
}

// RetireClaimRows archives the exact transaction snapshot before deletion.
// A missing or replacement row is successful only when this journal's committed
// archive proves the predecessor was already removed by an earlier attempt.
func (r *Registry) RetireClaimRows(ctx context.Context, journal string, targets []ClaimRow, archive func([]ClaimRowArchive) error) error {
	if archive == nil {
		return ErrClaimOwnership
	}
	if r.remote != nil {
		_, err := r.remote.call(ctx, remoteRequest{Method: "RetireClaimRows", Journal: journal, Targets: targets}, func(out remoteResponse) error { return archive(out.Archives) })
		return err
	}

	if journal == "" || len(targets) == 0 || archive == nil {
		return ErrClaimOwnership
	}
	tx, err := r.beginClaimTransaction(ctx, true)
	if err != nil {
		return err
	}
	defer tx.Rollback()
	archives := make([]ClaimRowArchive, 0, len(targets))
	pending := make([]bool, len(targets))
	seen := map[string]bool{}
	for i, t := range targets {
		if t.AllowAbsent && t.Kind != "timer" {
			return ErrClaimOwnership
		}
		if !t.Ownership.Valid() || t.Host == "" || (t.Kind != "route" && t.Kind != "timer") || (t.Kind == "timer" && (t.Name == "" || t.Registration == "")) || (t.Kind == "route" && (t.Name != "" || t.Registration != "")) {
			return ErrClaimOwnership
		}
		key := t.Kind + "\x00" + t.Host + "\x00" + t.Name
		if seen[key] {
			return ErrClaimOwnership
		}
		seen[key] = true
		var saved []byte
		err = tx.QueryRowContext(ctx, `SELECT archive FROM claim_row_retirements WHERE journal=? AND kind=? AND host=? AND name=? AND row_id=?`, journal, t.Kind, t.Host, t.Name, t.Ownership.ClaimRowID).Scan(&saved)
		if err == nil {
			var a ClaimRowArchive
			if json.Unmarshal(saved, &a) != nil || a.Row != t || len(a.Contents) == 0 {
				return ErrClaimOwnership
			}
			archives = append(archives, a)
			continue
		}
		if !errors.Is(err, sql.ErrNoRows) {
			return err
		}
		var content []byte
		switch t.Kind {
		case "route":
			row, e := scanRoute(tx.QueryRowContext(ctx, `SELECT `+routeCols+` FROM routes WHERE host=?`, t.Host).Scan)
			if e != nil {
				return fmt.Errorf("%w: %v", ErrClaimOwnership, e)
			}
			if row.ClaimOwnership != t.Ownership {
				return ErrClaimOwnership
			}
			content, err = json.Marshal(row)
		case "timer":
			row, e := scanTimer(tx.QueryRowContext(ctx, `SELECT `+timerCols+` FROM timers WHERE host=? AND name=?`, t.Host, t.Name))
			if errors.Is(e, sql.ErrNoRows) && t.AllowAbsent {
				content, err = []byte("null"), nil
				break
			}
			if e != nil {
				return fmt.Errorf("%w: %v", ErrClaimOwnership, e)
			}
			if row.ClaimOwnership != t.Ownership || row.Registration != t.Registration {
				return ErrClaimOwnership
			}
			content, err = json.Marshal(struct {
				Timer        Timer
				Registration string
			}{row, row.Registration})
		}
		if err != nil {
			return err
		}
		archives = append(archives, ClaimRowArchive{Row: t, Contents: content})
		pending[i] = true
	}
	if err = archive(archives); err != nil {
		return err
	}
	for i, t := range targets {
		if !pending[i] {
			continue
		}
		a, err := json.Marshal(archives[i])
		if err != nil {
			return err
		}
		if _, err = tx.ExecContext(ctx, `INSERT INTO claim_row_retirements (journal,kind,host,name,row_id,archive) VALUES (?,?,?,?,?,?)`, journal, t.Kind, t.Host, t.Name, t.Ownership.ClaimRowID, a); err != nil {
			return err
		}
		table := "routes"
		q := ""
		args := []any{t.Host, t.Ownership.ClaimIncarnation, t.Ownership.ClaimStamp, t.Ownership.ClaimOwner, t.Ownership.ClaimRowID}
		if t.Kind == "timer" {
			table = "timers"
			q = " AND name=? AND registration=?"
			args = append(args, t.Name, t.Registration)
		}
		res, e := tx.ExecContext(ctx, `DELETE FROM `+table+` WHERE host=? AND claim_incarnation=? AND claim_stamp=? AND claim_owner=? AND claim_row_id=?`+q, args...)
		if e != nil {
			return e
		}
		n, e := res.RowsAffected()
		if e != nil {
			return e
		}
		if n != 1 && !(n == 0 && t.AllowAbsent && string(archives[i].Contents) == "null") {
			return ErrClaimOwnership
		}
	}
	return tx.Commit()
}

func (r *Registry) AdoptClaimTimer(ctx context.Context, observed Timer, owner ClaimOwnership) error {
	if r.remote != nil {
		_, err := r.remote.call(ctx, remoteRequest{Method: "AdoptClaimTimer", ObservedValue: observed, Owner: owner}, nil)
		return err
	}

	if !owner.Valid() || observed.ClaimOwnership != (ClaimOwnership{}) || observed.Registration == "" || owner.ClaimRowID != observed.Registration {
		return ErrClaimOwnership
	}
	tx, err := r.beginClaimTransaction(ctx, true)
	if err != nil {
		return err
	}
	defer tx.Rollback()
	current, err := scanTimer(tx.QueryRowContext(ctx, `SELECT `+timerCols+` FROM timers WHERE host=? AND name=?`, observed.Host, observed.Name))
	if err != nil {
		return err
	}
	if current != observed {
		return ErrClaimOwnership
	}
	res, err := tx.ExecContext(ctx, `UPDATE timers SET claim_incarnation=?,claim_stamp=?,claim_owner=?,claim_row_id=? WHERE host=? AND name=? AND registration=? AND claim_owner='' AND claim_stamp=0 AND claim_incarnation='' AND claim_row_id=''`, owner.ClaimIncarnation, owner.ClaimStamp, owner.ClaimOwner, owner.ClaimRowID, observed.Host, observed.Name, observed.Registration)
	if err != nil {
		return err
	}
	n, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if n != 1 {
		return ErrClaimOwnership
	}
	return tx.Commit()
}