package cabana

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io/fs"
	"reflect"
	"slices"
	"strings"

	"git.golem15.com/golem15/summercms/modules/lagoon"
	"git.golem15.com/golem15/summercms/modules/pact"
	"git.golem15.com/golem15/summercms/modules/phrasebook"
	"github.com/goccy/go-yaml"
	"gocloud.dev/blob"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

// AdminRelationContractProvider declares model-owned relation metadata.
// Framework code never guesses a plugin pivot table or foreign key.
type AdminRelationContractProvider interface {
	AdminRelationContracts() []RelationContract
}

// Relation contract kinds (RelationContract.Kind).
const (
	// RelationBelongsToMany links related records through a pivot model.
	// It is the kind of a contract whose Kind is empty.
	RelationBelongsToMany = "belongsToMany"
	// RelationHasMany owns related records through their ForeignKey column.
	RelationHasMany = "hasMany"
)

// RelationContract binds one compiled relation schema to target and pivot models.
type RelationContract struct {
	Name string
	// Kind is RelationBelongsToMany or RelationHasMany. The empty value is
	// RelationBelongsToMany, so a contract written before hasMany existed
	// keeps its pivot behaviour unchanged.
	Kind       string
	NewRelated func() any
	// NewPivot, ParentForeignKey, RelatedForeignKey and HookPivotColumns
	// describe the pivot of a belongsToMany relation; a hasMany contract
	// leaves them empty.
	NewPivot          func() any
	ParentForeignKey  string
	RelatedForeignKey string
	// ForeignKey is the related model's column that points at the parent
	// (hasMany only). A pointer Go type (a nullable column) is needed for
	// unlink and for managing the relation before the parent is saved.
	ForeignKey         string
	Columns            map[string]string
	HookPivotColumns   []string
	ExcludedRelatedIDs func(parent any) ([]uint, error)
}

// kind is the contract's normalized kind.
func (c RelationContract) kind() string {
	if c.Kind == "" {
		return RelationBelongsToMany
	}
	return c.Kind
}

// RelationColumn is one source-ordered relation list column.
type RelationColumn struct {
	Key        string `json:"key"`
	Label      string `json:"label"`
	Searchable bool   `json:"searchable"`
	Sortable   bool   `json:"sortable"`
}

// RelationList is one view/manage list declaration.
type RelationList struct {
	Columns []RelationColumn `json:"columns"`
}

// RelationPanel is one side of a relation manager.
type RelationPanel struct {
	List           RelationList `json:"list"`
	ToolbarButtons []string     `json:"toolbarButtons"`
	ShowSearch     bool         `json:"showSearch"`
}

// RelationSchema is the cached locale-neutral relation contract.
type RelationSchema struct {
	Name  string `json:"name"`
	Label string `json:"label"`
	// Kind is the contract kind: belongsToMany or hasMany.
	Kind string `json:"kind"`
	// Deferrable is true when the relation can be managed on a record that
	// is not saved yet: always for belongsToMany, and for a hasMany whose
	// ForeignKey is nullable.
	Deferrable bool          `json:"deferrable"`
	View       RelationPanel `json:"view"`
	Manage     RelationPanel `json:"manage"`
	// ManageForm is the child create and update form (manage.form, or the
	// top-level form); ViewForm is the read-only preview form (view.form,
	// or the top-level form); PivotForm edits pivot columns of a
	// belongsToMany link (pivot.form). Each is omitted when not declared.
	ManageForm []FormField `json:"manageForm,omitempty"`
	ViewForm   []FormField `json:"viewForm,omitempty"`
	PivotForm  []FormField `json:"pivotForm,omitempty"`
	// Messages is the relation manager's copy (D-13); the cached schema
	// carries each phrase key as its own form.
	Messages *RelationMessages `json:"messages"`

	messageKeys relationMessageKeys
	// manageForm, viewForm and pivotForm are the compiled forms the field
	// lists come from; relatedModel and pivotModel supply dropdown options.
	manageForm   *FormSchema
	viewForm     *FormSchema
	pivotForm    *FormSchema
	relatedModel func() any
	pivotModel   func() any
}

// CompiledRelation combines trusted YAML with model-owned metadata.
type CompiledRelation struct {
	Schema              *RelationSchema
	Contract            RelationContract
	RequiredPermissions []string

	// kind is the normalized contract kind; deferrable mirrors
	// Schema.Deferrable; fieldName is the relation-manager form field.
	kind       string
	deferrable bool
	fieldName  string
	// child is the related model's form (the manage form) compiled as a
	// controller of its own: writable fields, file and date fields. view is
	// the read-only form when it differs. pivot is the pivot form bound to
	// the pivot model. Each is nil when not declared.
	child *CompiledController
	view  *CompiledController
	pivot *CompiledController
}

// hasMany reports whether the relation owns its children by foreign key.
func (cr *CompiledRelation) hasMany() bool { return cr != nil && cr.kind == RelationHasMany }

// allows reports whether the view panel declares the toolbar button.
func (cr *CompiledRelation) allows(button string) bool {
	return cr != nil && cr.Schema != nil && slices.Contains(cr.Schema.View.ToolbarButtons, button)
}

// RelationQuery is the finite linked/candidate query contract.
type RelationQuery struct {
	Search  string
	Sort    string
	Dir     string
	Page    string
	PerPage string
}

// RelationResult is a stable allocated page.
type RelationResult struct {
	Data []map[string]any
	Meta ListMeta
}

// RelationMutationInput is the only accepted relation write payload.
type RelationMutationInput struct {
	IDs []any `json:"ids"`
}

// RelationMutationResult reports inserted/deleted pivot rows.
type RelationMutationResult struct {
	Linked  int `json:"linked,omitempty"`
	Removed int `json:"removed,omitempty"`
}

// RelationService executes compiled relation reads and writes.
type RelationService struct {
	DB *gorm.DB

	// bucket deletes the blobs of child files a save removes, after commit;
	// tr localizes date bound messages. Both may be nil.
	bucket *blob.Bucket
	tr     *phrasebook.Translator
}

func (s RelationSchema) MarshalJSON() ([]byte, error) {
	if s.Messages == nil {
		keys := localizeMessages[relationMessageKeys, RelationMessages](context.Background(), nil, s.relationMessageKeySet())
		s.Messages = &keys
	}
	type alias RelationSchema
	out := alias(s)
	ensureRelationArrays(&out.View)
	ensureRelationArrays(&out.Manage)
	return json.Marshal(out)
}

func ensureRelationArrays(p *RelationPanel) {
	if p.List.Columns == nil {
		p.List.Columns = []RelationColumn{}
	}
	if p.ToolbarButtons == nil {
		p.ToolbarButtons = []string{}
	}
}

// Localize returns a request-local copy without mutating the cached keys.
func (s *RelationSchema) Localize(ctx context.Context, tr *phrasebook.Translator) *RelationSchema {
	if s == nil {
		return nil
	}
	out := *s
	out.Label = translateKey(ctx, tr, s.Label)
	out.View = localizeRelationPanel(ctx, tr, s.View)
	out.Manage = localizeRelationPanel(ctx, tr, s.Manage)
	messages := localizeMessages[relationMessageKeys, RelationMessages](ctx, tr, s.relationMessageKeySet())
	out.Messages = &messages
	out.ManageForm = localizeRelationForm(ctx, tr, s.manageForm, s.relatedModel, s.ManageForm)
	out.ViewForm = localizeRelationForm(ctx, tr, s.viewForm, s.relatedModel, s.ViewForm)
	out.PivotForm = localizeRelationForm(ctx, tr, s.pivotForm, s.pivotModel, s.PivotForm)
	return &out
}

// localizeRelationForm resolves a relation form's display strings, with the
// form's model as the dropdown options provider. A form that cannot be
// localized keeps its cached fields.
func localizeRelationForm(ctx context.Context, tr *phrasebook.Translator, form *FormSchema, model func() any, cached []FormField) []FormField {
	if form == nil {
		return cached
	}
	var provider pact.DropdownOptionsProvider
	if model != nil {
		provider = modelDropdownProvider(model())
	}
	view, err := form.Localize(ctx, tr, provider)
	if err != nil {
		return cached
	}
	return view.Fields
}

func localizeRelationPanel(ctx context.Context, tr *phrasebook.Translator, src RelationPanel) RelationPanel {
	out := src
	out.List.Columns = append([]RelationColumn(nil), src.List.Columns...)
	for i := range out.List.Columns {
		out.List.Columns[i].Label = translateKey(ctx, tr, out.List.Columns[i].Label)
	}
	out.ToolbarButtons = append([]string(nil), src.ToolbarButtons...)
	ensureRelationArrays(&out)
	return out
}

type relationRoot struct {
	Relations yaml.MapSlice `yaml:",inline"`
}

type relationDocument struct {
	Label string `yaml:"label"`
	// Form is WinterCMS's top-level form: the fallback of manage.form and
	// view.form.
	Form     string                 `yaml:"form"`
	View     relationPanelDocument  `yaml:"view"`
	Manage   relationPanelDocument  `yaml:"manage"`
	Pivot    *relationPivotDocument `yaml:"pivot"`
	Messages *relationMessageKeys   `yaml:"messages"`
}

type relationPanelDocument struct {
	List struct {
		Columns yaml.MapSlice `yaml:"columns"`
	} `yaml:"list"`
	Form           string `yaml:"form"`
	ToolbarButtons string `yaml:"toolbarButtons"`
	ShowSearch     bool   `yaml:"showSearch"`
}

type relationPivotDocument struct {
	Form string `yaml:"form"`
}

type relationColumnDocument struct {
	Label      string `yaml:"label"`
	Searchable *bool  `yaml:"searchable"`
	Sortable   *bool  `yaml:"sortable"`
}

func compileRelations(pluginID string, ctl pact.AdminController, fsys fs.FS, form *FormSchema) (map[string]*CompiledRelation, error) {
	// fields maps each relation-manager relation to its form field index.
	fields := map[string]int{}
	if form != nil {
		for i, field := range form.Fields {
			if field.Type == "relation-manager" {
				fields[field.Relation] = i
			}
		}
	}
	if len(fields) == 0 {
		return map[string]*CompiledRelation{}, nil
	}
	provider, ok := ctl.(AdminRelationContractProvider)
	if !ok || provider == nil {
		return nil, bootErr(pluginID, ctl.ID(), "config_relation.yaml", errors.New("relation-manager requires AdminRelationContracts"))
	}
	contracts := map[string]RelationContract{}
	for _, contract := range provider.AdminRelationContracts() {
		if !identifier(contract.Name) {
			return nil, bootErr(pluginID, ctl.ID(), "config_relation.yaml", fmt.Errorf("relation contract name %q is not an identifier", contract.Name))
		}
		if _, dup := contracts[contract.Name]; dup {
			return nil, bootErr(pluginID, ctl.ID(), "config_relation.yaml", fmt.Errorf("duplicate relation contract %s", contract.Name))
		}
		if err := validateRelationContract(ctl, contract); err != nil {
			return nil, bootErr(pluginID, ctl.ID(), "config_relation.yaml", err)
		}
		contracts[contract.Name] = contract
	}
	dir := strings.Trim(strings.TrimSpace(ctl.ConfigDir()), "/")
	file := dir + "/config_relation.yaml"
	raw, err := readAsset(fsys, file)
	if err != nil {
		return nil, bootErr(pluginID, ctl.ID(), file, err)
	}
	var root yaml.MapSlice
	if err := decodeStrictOrdered(raw, &root); err != nil {
		return nil, bootErr(pluginID, ctl.ID(), file, err)
	}
	out := map[string]*CompiledRelation{}
	for _, item := range root {
		name, ok := item.Key.(string)
		if !ok || !identifier(name) {
			return nil, bootErr(pluginID, ctl.ID(), file, fmt.Errorf("relation name %v is not an identifier", item.Key))
		}
		if _, dup := out[name]; dup {
			return nil, bootErr(pluginID, ctl.ID(), file, fmt.Errorf("duplicate relation %s", name))
		}
		if _, declared := fields[name]; !declared {
			return nil, bootErr(pluginID, ctl.ID(), file, fmt.Errorf("relation %s has no relation-manager field", name))
		}
		contract, declared := contracts[name]
		if !declared {
			return nil, bootErr(pluginID, ctl.ID(), file, fmt.Errorf("relation %s has no model contract", name))
		}
		encoded, err := yaml.Marshal(item.Value)
		if err != nil {
			return nil, bootErr(pluginID, ctl.ID(), file, err)
		}
		var doc relationDocument
		if err := decodeStrict(encoded, &doc); err != nil {
			return nil, bootErr(pluginID, ctl.ID(), file, fmt.Errorf("relation %s: %w", name, err))
		}
		view, err := compileRelationPanel(doc.View, contract, true)
		if err != nil {
			return nil, bootErr(pluginID, ctl.ID(), file, fmt.Errorf("relation %s view: %w", name, err))
		}
		manage, err := compileRelationPanel(doc.Manage, contract, false)
		if err != nil {
			return nil, bootErr(pluginID, ctl.ID(), file, fmt.Errorf("relation %s manage: %w", name, err))
		}
		schema := &RelationSchema{Name: name, Label: doc.Label, Kind: contract.kind(), View: view, Manage: manage}
		if doc.Messages != nil {
			schema.messageKeys = *doc.Messages
		}
		keys := localizeMessages[relationMessageKeys, RelationMessages](context.Background(), nil, schema.relationMessageKeySet())
		schema.Messages = &keys
		cr := &CompiledRelation{
			Schema:              schema,
			Contract:            contract,
			RequiredPermissions: append([]string(nil), requiredOf(ctl)...),
			kind:                contract.kind(),
			deferrable:          relationDeferrable(contract),
			fieldName:           form.Fields[fields[name]].Name,
		}
		schema.Deferrable = cr.deferrable
		if err := compileRelationForms(pluginID, ctl, fsys, file, doc, cr); err != nil {
			return nil, err
		}
		form.Fields[fields[name]].Deferrable = cr.deferrable
		out[name] = cr
	}
	for name := range fields {
		if out[name] == nil {
			return nil, bootErr(pluginID, ctl.ID(), file, fmt.Errorf("relation %s is missing", name))
		}
	}
	for name := range contracts {
		if out[name] == nil {
			return nil, bootErr(pluginID, ctl.ID(), file, fmt.Errorf("relation contract %s is orphaned", name))
		}
	}
	return out, nil
}

func compileRelationPanel(doc relationPanelDocument, contract RelationContract, view bool) (RelationPanel, error) {
	cols := make([]RelationColumn, 0, len(doc.List.Columns))
	seen := map[string]struct{}{}
	for _, item := range doc.List.Columns {
		key, ok := item.Key.(string)
		if !ok || !identifier(key) {
			return RelationPanel{}, fmt.Errorf("column %v is not an identifier", item.Key)
		}
		if _, dup := seen[key]; dup {
			return RelationPanel{}, fmt.Errorf("duplicate column %s", key)
		}
		seen[key] = struct{}{}
		if _, ok := contract.Columns[key]; !ok {
			return RelationPanel{}, fmt.Errorf("column %s is not in the model contract", key)
		}
		encoded, err := yaml.Marshal(item.Value)
		if err != nil {
			return RelationPanel{}, err
		}
		var spec relationColumnDocument
		if err := decodeStrict(encoded, &spec); err != nil {
			return RelationPanel{}, fmt.Errorf("column %s: %w", key, err)
		}
		searchable := doc.ShowSearch
		if spec.Searchable != nil {
			searchable = *spec.Searchable
		}
		sortable := true
		if spec.Sortable != nil {
			sortable = *spec.Sortable
		}
		cols = append(cols, RelationColumn{Key: key, Label: spec.Label, Searchable: searchable, Sortable: sortable})
	}
	buttons, err := compileRelationButtons(doc.ToolbarButtons, view)
	if err != nil {
		return RelationPanel{}, err
	}
	return RelationPanel{List: RelationList{Columns: cols}, ToolbarButtons: buttons, ShowSearch: doc.ShowSearch}, nil
}

// relationButtons are the WinterCMS RelationController toolbar buttons the
// view panel may declare (D-12). Each one is the capability of its routes:
// create, update (row edit), delete, link and unlink.
var relationButtons = []string{"create", "update", "delete", "link", "unlink"}

func compileRelationButtons(raw string, view bool) ([]string, error) {
	if strings.TrimSpace(raw) == "" {
		return []string{}, nil
	}
	parts := strings.Split(raw, "|")
	out := make([]string, 0, len(parts))
	seen := map[string]struct{}{}
	for _, part := range parts {
		part = strings.TrimSpace(part)
		if !slices.Contains(relationButtons, part) {
			return nil, fmt.Errorf("unsupported relation action %s", part)
		}
		if !view && part != "link" {
			return nil, fmt.Errorf("manage panel cannot declare %s (only link)", part)
		}
		if _, dup := seen[part]; dup {
			return nil, fmt.Errorf("duplicate relation action %s", part)
		}
		seen[part] = struct{}{}
		out = append(out, part)
	}
	return out, nil
}

func validateRelationContract(ctl pact.AdminController, contract RelationContract) error {
	switch contract.kind() {
	case RelationBelongsToMany:
		if err := validatePivotContract(contract); err != nil {
			return err
		}
	case RelationHasMany:
		if err := validateHasManyContract(contract); err != nil {
			return err
		}
	default:
		return fmt.Errorf("relation %s has unknown kind %q (want %s or %s)", contract.Name, contract.Kind, RelationBelongsToMany, RelationHasMany)
	}
	targetCols := modelColumns(contract.NewRelated())
	for logical, physical := range contract.Columns {
		if !identifier(logical) || !identifier(physical) {
			return fmt.Errorf("relation %s has invalid column mapping", contract.Name)
		}
		if _, ok := targetCols[physical]; !ok {
			return fmt.Errorf("relation %s target is missing column %s", contract.Name, physical)
		}
	}
	src, ok := ctl.(pact.AdminRecordSource)
	if !ok || src == nil || src.NewRecord() == nil {
		return fmt.Errorf("relation %s owner has no record source", contract.Name)
	}
	fieldName, ok := relationFieldName(src.NewRecord(), contract.Name)
	if !ok {
		return fmt.Errorf("relation %s is not on the owner model", contract.Name)
	}
	ownerType := reflect.TypeOf(src.NewRecord())
	for ownerType.Kind() == reflect.Pointer {
		ownerType = ownerType.Elem()
	}
	field, _ := ownerType.FieldByName(fieldName)
	targetType := reflect.TypeOf(contract.NewRelated())
	for targetType.Kind() == reflect.Pointer {
		targetType = targetType.Elem()
	}
	fieldType := field.Type
	for fieldType.Kind() == reflect.Pointer || fieldType.Kind() == reflect.Slice {
		fieldType = fieldType.Elem()
	}
	if fieldType != targetType {
		return fmt.Errorf("relation %s target model does not match owner field", contract.Name)
	}
	return nil
}

// validatePivotContract checks a belongsToMany contract: target and pivot
// models, both pivot foreign keys on the pivot model and the hook columns.
func validatePivotContract(contract RelationContract) error {
	if contract.NewRelated == nil || contract.NewRelated() == nil || contract.NewPivot == nil || contract.NewPivot() == nil {
		return fmt.Errorf("relation %s requires target and pivot models", contract.Name)
	}
	if contract.ForeignKey != "" {
		return fmt.Errorf("relation %s: a belongsToMany contract cannot declare ForeignKey", contract.Name)
	}
	if !identifier(contract.ParentForeignKey) || !identifier(contract.RelatedForeignKey) {
		return fmt.Errorf("relation %s has invalid pivot foreign keys", contract.Name)
	}
	pivotCols := modelColumns(contract.NewPivot())
	if _, ok := pivotCols[contract.ParentForeignKey]; !ok {
		return fmt.Errorf("relation %s pivot is missing %s", contract.Name, contract.ParentForeignKey)
	}
	if _, ok := pivotCols[contract.RelatedForeignKey]; !ok {
		return fmt.Errorf("relation %s pivot is missing %s", contract.Name, contract.RelatedForeignKey)
	}
	for _, col := range contract.HookPivotColumns {
		if !identifier(col) {
			return fmt.Errorf("relation %s has invalid hook column %s", contract.Name, col)
		}
		if _, ok := pivotCols[col]; !ok || protectedPivotColumn(col, contract) {
			return fmt.Errorf("relation %s has invalid hook column %s", contract.Name, col)
		}
	}
	return nil
}

// validateHasManyContract checks a hasMany contract: no pivot fields, and a
// ForeignKey that is an integer column (or a pointer to one) of the related
// model.
func validateHasManyContract(contract RelationContract) error {
	if contract.NewRelated == nil || contract.NewRelated() == nil {
		return fmt.Errorf("relation %s requires a target model", contract.Name)
	}
	if contract.NewPivot != nil || contract.ParentForeignKey != "" || contract.RelatedForeignKey != "" || len(contract.HookPivotColumns) > 0 {
		return fmt.Errorf("relation %s: a hasMany contract cannot declare NewPivot, ParentForeignKey, RelatedForeignKey or HookPivotColumns", contract.Name)
	}
	if !identifier(contract.ForeignKey) {
		return fmt.Errorf("relation %s: hasMany needs a ForeignKey column", contract.Name)
	}
	field, ok := structFieldByColumn(contract.NewRelated(), contract.ForeignKey)
	if !ok {
		return fmt.Errorf("relation %s target is missing column %s", contract.Name, contract.ForeignKey)
	}
	if !uintLike(field.Type) {
		return fmt.Errorf("relation %s: ForeignKey %s must be an integer column, found %s", contract.Name, contract.ForeignKey, field.Type)
	}
	return nil
}

// relationDeferrable reports whether a relation can be managed before its
// parent is saved: a belongsToMany always (the pivot row is written on
// save), a hasMany only with a nullable ForeignKey (the child is inserted
// with a NULL key first, as in WinterCMS).
func relationDeferrable(contract RelationContract) bool {
	if contract.kind() != RelationHasMany {
		return true
	}
	field, ok := structFieldByColumn(contract.NewRelated(), contract.ForeignKey)
	return ok && field.Type.Kind() == reflect.Pointer
}

func protectedPivotColumn(col string, contract RelationContract) bool {
	switch col {
	case contract.ParentForeignKey, contract.RelatedForeignKey, "id", "created_at", "updated_at", "deleted_at":
		return true
	default:
		return false
	}
}

// Linked returns the current relation rows.
func (s RelationService) Linked(ctx context.Context, cc *CompiledController, relation string, ownerID uint, in RelationQuery) (*RelationResult, error) {
	return s.query(ctx, cc, relation, ownerID, in, false)
}

// Candidates returns target rows eligible for linking.
func (s RelationService) Candidates(ctx context.Context, cc *CompiledController, relation string, ownerID uint, in RelationQuery) (*RelationResult, error) {
	return s.query(ctx, cc, relation, ownerID, in, true)
}

func (s RelationService) query(ctx context.Context, cc *CompiledController, relation string, ownerID uint, in RelationQuery, candidates bool) (*RelationResult, error) {
	if s.DB == nil {
		return nil, errors.New("cabana: database is not configured")
	}
	cr, err := relationOf(cc, relation)
	if err != nil {
		return nil, err
	}
	page, per, sortKey, desc, err := normalizeRelationQuery(cr.Schema, candidates, in)
	if err != nil {
		return nil, err
	}
	var result *RelationResult
	err = s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		parent, err := newWritableModel(cc)
		if err != nil {
			return err
		}
		if err := loadRecord(ctx, tx, cc, parent, ownerID); err != nil {
			return err
		}
		q, target, err := relationBaseQuery(ctx, tx, cc, cr, parent, candidates)
		if err != nil {
			return err
		}
		panel := cr.Schema.View
		if candidates {
			panel = cr.Schema.Manage
		}
		q = applyRelationSearch(q, cr, panel.List.Columns, in.Search)
		var total int64
		if err := q.Session(&gorm.Session{}).Count(&total).Error; err != nil {
			return err
		}
		q = applyRelationOrder(q, cr, target, panel.List.Columns, sortKey, desc)
		rows := make([]map[string]any, 0)
		selects := relationSelects(tx, cr, target, panel.List.Columns)
		if err := q.Select(strings.Join(selects, ", ")).Offset((page - 1) * per).Limit(per).Scan(&rows).Error; err != nil {
			return err
		}
		last := 0
		if total > 0 {
			last = int((total + int64(per) - 1) / int64(per))
		}
		result = &RelationResult{Data: rows, Meta: ListMeta{Page: page, PerPage: per, Total: total, LastPage: last}}
		return nil
	})
	return result, err
}

func relationBaseQuery(ctx context.Context, tx *gorm.DB, cc *CompiledController, cr *CompiledRelation, parent any, candidates bool) (*gorm.DB, any, error) {
	if cr.hasMany() {
		return hasManyBaseQuery(ctx, tx, cc, cr, parent, candidates)
	}
	target := cr.Contract.NewRelated()
	targetTable := tableName(target)
	pivotTable := tableName(cr.Contract.NewPivot())
	pk := primaryColumn(target)
	ownerPK := pkUint(parent)
	q := tx.WithContext(ctx).Model(target)
	if candidates {
		if ext, ok := cc.Controller.(pact.RelationExtendManageQuery); ok && ext != nil {
			if next := ext.RelationExtendManageQuery(ctx, cr.Contract.Name, q); next != nil {
				q = next
			}
		}
		if cr.Contract.ExcludedRelatedIDs != nil {
			excluded, err := cr.Contract.ExcludedRelatedIDs(parent)
			if err != nil {
				return nil, nil, lifecycleFailure(cc, err)
			}
			if len(excluded) > 0 {
				q = q.Where(clause.Not(clause.IN{Column: clause.Column{Table: targetTable, Name: pk}, Values: uintValues(excluded)}))
			}
		}
		notExists := "NOT EXISTS (SELECT 1 FROM " + quotedIdent(tx, pivotTable) + " p WHERE p." + quotedIdent(tx, cr.Contract.ParentForeignKey) + " = ? AND p." + quotedIdent(tx, cr.Contract.RelatedForeignKey) + " = " + quotedIdent(tx, targetTable) + "." + quotedIdent(tx, pk) + ")"
		q = q.Where(notExists, ownerPK)
	} else {
		join := "JOIN " + quotedIdent(tx, pivotTable) + " p ON p." + quotedIdent(tx, cr.Contract.RelatedForeignKey) + " = " + quotedIdent(tx, targetTable) + "." + quotedIdent(tx, pk)
		q = q.Joins(join).Where("p."+quotedIdent(tx, cr.Contract.ParentForeignKey)+" = ?", ownerPK)
	}
	return q, target, nil
}

// hasManyBaseQuery is relationBaseQuery for a hasMany relation: the linked
// rows are the related rows whose ForeignKey is the parent's key.
func hasManyBaseQuery(ctx context.Context, tx *gorm.DB, cc *CompiledController, cr *CompiledRelation, parent any, candidates bool) (*gorm.DB, any, error) {
	target := cr.Contract.NewRelated()
	fk := clause.Column{Table: tableName(target), Name: cr.Contract.ForeignKey}
	q := tx.WithContext(ctx).Model(target)
	if !candidates {
		return q.Where(clause.Eq{Column: fk, Value: pkUint(parent)}), target, nil
	}
	// Candidates are rows no parent owns yet: a NULL ForeignKey.
	if ext, ok := cc.Controller.(pact.RelationExtendManageQuery); ok && ext != nil {
		if next := ext.RelationExtendManageQuery(ctx, cr.Contract.Name, q); next != nil {
			q = next
		}
	}
	if cr.Contract.ExcludedRelatedIDs != nil {
		excluded, err := cr.Contract.ExcludedRelatedIDs(parent)
		if err != nil {
			return nil, nil, lifecycleFailure(cc, err)
		}
		if len(excluded) > 0 {
			q = q.Where(clause.Not(clause.IN{Column: clause.Column{Table: tableName(target), Name: primaryColumn(target)}, Values: uintValues(excluded)}))
		}
	}
	return q.Where(clause.Eq{Column: fk, Value: nil}), target, nil
}

// normalizeRelationPage applies the Phase 9 relation paging limits: page is
// a positive integer (default 1), per_page is 1..100 (default 20).
func normalizeRelationPage(rawPage, rawPerPage string) (int, int, error) {
	page, per := 1, 20
	var err error
	if rawPage != "" {
		page, err = parsePositive(rawPage)
		if err != nil {
			return 0, 0, relationInvalid("page", "must be a positive integer")
		}
	}
	if rawPerPage != "" {
		per, err = parsePositive(rawPerPage)
		if err != nil || per > 100 {
			return 0, 0, relationInvalid("per_page", "must be between 1 and 100")
		}
	}
	return page, per, nil
}

func normalizeRelationQuery(schema *RelationSchema, candidates bool, in RelationQuery) (int, int, string, bool, error) {
	page, per, err := normalizeRelationPage(in.Page, in.PerPage)
	if err != nil {
		return 0, 0, "", false, err
	}
	panel := schema.View
	if candidates {
		panel = schema.Manage
	}
	sortKey := in.Sort
	if sortKey == "" {
		// No explicit sort: use the first sortable column, or none (the
		// primary key then orders the page). Only a sort the client names is
		// validated, so a panel whose first column is not sortable still loads.
		for _, col := range panel.List.Columns {
			if col.Sortable {
				sortKey = col.Key
				break
			}
		}
	} else {
		valid := false
		for _, col := range panel.List.Columns {
			if col.Key == sortKey && col.Sortable {
				valid = true
				break
			}
		}
		if !valid {
			return 0, 0, "", false, relationInvalid("sort", "is not a sortable column")
		}
	}
	dir := in.Dir
	if dir == "" {
		dir = "asc"
	}
	if dir != "asc" && dir != "desc" {
		return 0, 0, "", false, relationInvalid("dir", "must be asc or desc")
	}
	return page, per, sortKey, dir == "desc", nil
}

func parsePositive(raw string) (int, error) {
	n := 0
	for _, r := range raw {
		if r < '0' || r > '9' {
			return 0, errors.New("not positive")
		}
		n = n*10 + int(r-'0')
	}
	if n < 1 {
		return 0, errors.New("not positive")
	}
	return n, nil
}

func relationInvalid(field, message string) *ValidationError {
	return &ValidationError{Details: map[string]any{field: []string{message}}}
}

func applyRelationSearch(q *gorm.DB, cr *CompiledRelation, cols []RelationColumn, term string) *gorm.DB {
	term = strings.TrimSpace(term)
	if term == "" {
		return q
	}
	targetTable := tableName(cr.Contract.NewRelated())
	parts := make([]string, 0)
	args := make([]any, 0)
	seen := map[string]struct{}{}
	for _, col := range cols {
		physical := cr.Contract.Columns[col.Key]
		if !col.Searchable || physical == "" {
			continue
		}
		if _, dup := seen[physical]; dup {
			continue
		}
		seen[physical] = struct{}{}
		parts = append(parts, quotedIdent(q, targetTable)+"."+quotedIdent(q, physical)+" ILIKE ?")
		args = append(args, "%"+term+"%")
	}
	if len(parts) == 0 {
		return q
	}
	return q.Where("("+strings.Join(parts, " OR ")+")", args...)
}

func applyRelationOrder(q *gorm.DB, cr *CompiledRelation, target any, cols []RelationColumn, sortKey string, desc bool) *gorm.DB {
	table := tableName(target)
	pk := primaryColumn(target)
	if sortKey != "" {
		q = q.Order(clause.OrderByColumn{Column: clause.Column{Table: table, Name: cr.Contract.Columns[sortKey]}, Desc: desc})
	}
	if cr.Contract.Columns[sortKey] != pk {
		q = q.Order(clause.OrderByColumn{Column: clause.Column{Table: table, Name: pk}})
	}
	return q
}

func relationSelects(db *gorm.DB, cr *CompiledRelation, target any, cols []RelationColumn) []string {
	table := tableName(target)
	out := []string{quotedIdent(db, table) + "." + quotedIdent(db, primaryColumn(target)) + " AS " + quotedIdent(db, "id")}
	for _, col := range cols {
		out = append(out, quotedIdent(db, table)+"."+quotedIdent(db, cr.Contract.Columns[col.Key])+" AS "+quotedIdent(db, col.Key))
	}
	return out
}

// Link inserts only currently eligible targets and never restamps existing rows.
func (s RelationService) Link(ctx context.Context, cc *CompiledController, relation string, ownerID uint, in RelationMutationInput) (RelationMutationResult, error) {
	ids, err := normalizeIDs(in.IDs)
	if err != nil {
		return RelationMutationResult{}, err
	}
	cr, err := relationOf(cc, relation)
	if err != nil {
		return RelationMutationResult{}, err
	}
	if cr.hasMany() {
		return RelationMutationResult{}, recordNotFound{}
	}
	var result RelationMutationResult
	err = lagoon.Transaction(ctx, s.DB, func(ctx context.Context, tx *gorm.DB) error {
		ctx = withTx(ctx, tx)
		parent, err := newWritableModel(cc)
		if err != nil {
			return err
		}
		if err := loadRecord(ctx, tx, cc, parent, ownerID); err != nil {
			return err
		}
		ownerPK := pkUint(parent)
		pending, err := pendingRelationIDs(tx, cr, ownerPK, ids)
		if err != nil || len(pending) == 0 {
			return err
		}
		q, target, err := relationBaseQuery(ctx, tx, cc, cr, parent, true)
		if err != nil {
			return err
		}
		q = q.Clauses(clause.Locking{Strength: "UPDATE"}).Where(clause.IN{Column: clause.Column{Table: tableName(target), Name: primaryColumn(target)}, Values: uintValues(pending)})
		holder := reflect.New(reflect.SliceOf(reflect.TypeOf(target).Elem()))
		if err := q.Order(clause.OrderByColumn{Column: clause.Column{Table: tableName(target), Name: primaryColumn(target)}}).Find(holder.Interface()).Error; err != nil {
			return err
		}
		if holder.Elem().Len() != len(pending) {
			return relationInvalid("ids", "contains an ineligible target")
		}
		for i := 0; i < holder.Elem().Len(); i++ {
			related := holder.Elem().Index(i).Addr().Interface()
			if err := insertPivot(ctx, tx, cc, cr, parent, related, nil); err != nil {
				return err
			}
			result.Linked++
		}
		return nil
	})
	return result, err
}

// insertPivot writes the pivot row linking related to the saved parent of a
// belongsToMany relation. A filled pivot model (from the pivot form) may be
// passed; RelationBeforeLink then stamps only its HookPivotColumns, and the
// two foreign keys are always set here.
func insertPivot(ctx context.Context, tx *gorm.DB, cc *CompiledController, cr *CompiledRelation, parent, related, pivot any) error {
	if pivot == nil {
		pivot = cr.Contract.NewPivot()
	}
	if err := setModelColumn(pivot, cr.Contract.ParentForeignKey, pkUint(parent)); err != nil {
		return err
	}
	if err := setModelColumn(pivot, cr.Contract.RelatedForeignKey, pkUint(related)); err != nil {
		return err
	}
	values := map[string]any{}
	if hook, ok := cc.Controller.(pact.RelationBeforeLink); ok && hook != nil {
		if err := hook.RelationBeforeLink(ctx, cr.Contract.Name, parent, related, values); err != nil {
			return lifecycleFailure(cc, err)
		}
	}
	for key, value := range values {
		if !stringSliceHasValue(cr.Contract.HookPivotColumns, key) || protectedPivotColumn(key, cr.Contract) {
			return lifecycleFailure(cc, fmt.Errorf("relation hook wrote protected pivot column"))
		}
		if err := setModelColumn(pivot, key, value); err != nil {
			return lifecycleFailure(cc, err)
		}
	}
	return tx.WithContext(ctx).Create(pivot).Error
}

// Unlink deletes explicit pivot models so their lifecycle hooks run.
func (s RelationService) Unlink(ctx context.Context, cc *CompiledController, relation string, ownerID uint, in RelationMutationInput) (RelationMutationResult, error) {
	ids, err := normalizeIDs(in.IDs)
	if err != nil {
		return RelationMutationResult{}, err
	}
	cr, err := relationOf(cc, relation)
	if err != nil {
		return RelationMutationResult{}, err
	}
	if cr.hasMany() {
		return RelationMutationResult{}, recordNotFound{}
	}
	var result RelationMutationResult
	err = lagoon.Transaction(ctx, s.DB, func(ctx context.Context, tx *gorm.DB) error {
		ctx = withTx(ctx, tx)
		parent, err := newWritableModel(cc)
		if err != nil {
			return err
		}
		if err := loadRecord(ctx, tx, cc, parent, ownerID); err != nil {
			return err
		}
		proto := cr.Contract.NewPivot()
		t := reflect.TypeOf(proto)
		holder := reflect.New(reflect.SliceOf(t.Elem()))
		q := tx.WithContext(ctx).Model(proto).Clauses(clause.Locking{Strength: "UPDATE"}).
			Where(clause.Eq{Column: clause.Column{Name: cr.Contract.ParentForeignKey}, Value: pkUint(parent)}).
			Where(clause.IN{Column: clause.Column{Name: cr.Contract.RelatedForeignKey}, Values: uintValues(ids)}).
			Order(clause.OrderByColumn{Column: clause.Column{Name: cr.Contract.RelatedForeignKey}})
		if err := q.Find(holder.Interface()).Error; err != nil {
			return err
		}
		for i := 0; i < holder.Elem().Len(); i++ {
			if err := tx.WithContext(ctx).Delete(holder.Elem().Index(i).Addr().Interface()).Error; err != nil {
				return err
			}
			result.Removed++
		}
		return nil
	})
	return result, err
}

func relationOf(cc *CompiledController, name string) (*CompiledRelation, error) {
	if cc == nil || !identifier(name) || cc.Relations == nil || cc.Relations[name] == nil {
		return nil, recordNotFound{}
	}
	return cc.Relations[name], nil
}

func pendingRelationIDs(tx *gorm.DB, cr *CompiledRelation, ownerID uint, ids []uint) ([]uint, error) {
	var existing []uint
	err := tx.Table(tableName(cr.Contract.NewPivot())).
		Where(clause.Eq{Column: clause.Column{Name: cr.Contract.ParentForeignKey}, Value: ownerID}).
		Where(clause.IN{Column: clause.Column{Name: cr.Contract.RelatedForeignKey}, Values: uintValues(ids)}).
		Pluck(cr.Contract.RelatedForeignKey, &existing).Error
	if err != nil {
		return nil, err
	}
	seen := map[uint]struct{}{}
	for _, id := range existing {
		seen[id] = struct{}{}
	}
	pending := make([]uint, 0, len(ids)-len(seen))
	for _, id := range ids {
		if _, ok := seen[id]; !ok {
			pending = append(pending, id)
		}
	}
	return pending, nil
}

func uintValues(ids []uint) []any {
	out := make([]any, len(ids))
	for i, id := range ids {
		out[i] = id
	}
	return out
}

func stringSliceHasValue(values []string, key string) bool {
	for _, value := range values {
		if value == key {
			return true
		}
	}
	return false
}

func setModelColumn(model any, column string, value any) error {
	v := reflect.ValueOf(model)
	if v.Kind() != reflect.Pointer || v.IsNil() {
		return errors.New("cabana: pivot model must be a pointer")
	}
	v = v.Elem()
	if v.Kind() != reflect.Struct {
		return errors.New("cabana: pivot model must be a struct pointer")
	}
	field := fieldByColumn(v, column)
	if !field.IsValid() || !field.CanSet() {
		return fmt.Errorf("cabana: pivot column %s is not writable", column)
	}
	return assignReflect(field, value)
}

func assignReflect(dst reflect.Value, value any) error {
	if value == nil {
		if dst.Kind() == reflect.Pointer {
			dst.Set(reflect.Zero(dst.Type()))
			return nil
		}
		return errors.New("cabana: nil pivot value")
	}
	src := reflect.ValueOf(value)
	if src.Type().AssignableTo(dst.Type()) {
		dst.Set(src)
		return nil
	}
	if dst.Kind() == reflect.Pointer {
		if src.Kind() == reflect.Pointer && src.Type().AssignableTo(dst.Type()) {
			dst.Set(src)
			return nil
		}
		ptr := reflect.New(dst.Type().Elem())
		if src.Type().AssignableTo(dst.Type().Elem()) {
			ptr.Elem().Set(src)
			dst.Set(ptr)
			return nil
		}
		if src.Type().ConvertibleTo(dst.Type().Elem()) {
			ptr.Elem().Set(src.Convert(dst.Type().Elem()))
			dst.Set(ptr)
			return nil
		}
	}
	if src.Type().ConvertibleTo(dst.Type()) {
		dst.Set(src.Convert(dst.Type()))
		return nil
	}
	return fmt.Errorf("cabana: pivot value %T is not assignable to %s", value, dst.Type())
}