package console

// claim_sql_grants.go is the only production door into the SQL gate
// (store_gate.go). A grant is one app incarnation plus the canonical
// authority that must still hold before every write and every commit inside
// the callback. Grants are built here, from canonical admission and the
// bucket, by unexported constructors; there is no grant that means "trust the
// caller".
//
// Each grant binds one permit mode (grantModes): effect-time permits may
// write anything of their incarnation; the terminal replay grants (restore
// receipt, trash marker, purge receipt) bind restore, trash and purge
// permits, whose statement families SQLite's triggers enforce row by row.
//
// WithAppMutation lends the callback a Store view bound to one held
// connection. The callback must use only that view: the outer store's pool is
// the held connection, so an outer call deadlocks until its context ends, and
// it would bypass the permits anyway. Keep network calls out of callbacks;
// they hold the console's only SQLite connection.

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"slices"

	"github.com/carlosframework/platform/internal/hostops"
	"github.com/carlosframework/platform/internal/release"
)

// sqlGrant is one incarnation a WithAppMutation callback may write, and the
// authority that must keep holding while it does.
type sqlGrant struct {
	scope    claimSQLScope
	kind     string
	check    func(context.Context) error
	local    func(context.Context) error
	prepared **hostops.PreparedSQLPermit
}

// WithAppMutation runs fn on a view of s that may write exactly the granted
// incarnations' rows, re-validating every grant before each write and commit.
//
// On an ungated store (Open: flag-off, and package tests) there are no
// triggers to bind, so the grants are validated once and fn runs on s. The
// flag-off console never calls this: NewServer refuses HostClaim on an
// ungated store, and only claim code builds grants.
func (s *Store) WithAppMutation(ctx context.Context, grants []sqlGrant, fn func(*Store) error) (err error) {
	if fn == nil || len(grants) == 0 {
		return fmt.Errorf("%w: grants and fn are required", ErrClaimSQLScope)
	}
	// Refuse nesting before any remote preparation runs.
	if s.bound != nil {
		return fmt.Errorf("%w: nested WithAppMutation", ErrClaimSQLScope)
	}
	if s.outerInsideView() {
		return errOuterStoreInView
	}
	byScope := map[claimSQLScope][]sqlGrant{}
	scopes := make([]claimSQLScope, 0, len(grants))
	for _, g := range grants {
		if g.check == nil || g.kind == "" || g.scope.AccountID <= 0 || g.scope.App == "" || g.scope.AppID == "" {
			return fmt.Errorf("%w: a grant needs an account, app, app_id, kind and check", ErrClaimSQLScope)
		}
		if mode, ok := grantModes[g.kind]; !ok || g.scope.Mode != mode {
			return fmt.Errorf("%w: grant kind %q cannot bind permit mode %q", ErrClaimSQLScope, g.kind, g.scope.Mode)
		}
		if _, seen := byScope[g.scope]; !seen {
			scopes = append(scopes, g.scope)
		}
		byScope[g.scope] = append(byScope[g.scope], g)
	}
	check := func(ctx context.Context, sc claimSQLScope) error {
		gs, ok := byScope[sc]
		if !ok {
			return ErrClaimSQLScope
		}
		for _, g := range gs {
			if err := g.check(ctx); err != nil {
				return fmt.Errorf("%s grant: %w", g.kind, err)
			}
		}
		return nil
	}
	if s.gate == nil {
		for _, sc := range scopes {
			if err := check(ctx, sc); err != nil {
				return fmt.Errorf("%w: %w", ErrClaimSQLDenied, err)
			}
		}
		return fn(s)
	}
	local := func(ctx context.Context, sc claimSQLScope) error {
		for _, g := range byScope[sc] {
			if g.local != nil {
				if err := g.local(ctx); err != nil {
					return err
				}
			}
		}
		return ctx.Err()
	}
	// Every prepared permit that names a receipt effect is finished on every
	// path, including a refusal before the connection is bound: the local
	// fence proves at once that nothing ran, so the effect does not wait out
	// its registration.
	var permits []*hostops.PreparedSQLPermit
	collect := func() {
		for _, g := range grants {
			if g.prepared != nil && *g.prepared != nil && !slices.Contains(permits, *g.prepared) {
				permits = append(permits, *g.prepared)
			}
		}
	}
	defer func() {
		if value := recover(); value != nil {
			for _, p := range permits {
				if p.Receipt().Effect != "" {
					_ = p.FinishTransaction(errors.New("SQL callback panicked"))
				}
			}
			panic(value)
		}
		for i := len(permits) - 1; i >= 0; i-- {
			if p := permits[i]; p.Receipt().Effect != "" {
				err = errors.Join(err, s.finishPreparedSQL(ctx, p, err))
			}
		}
	}()
	// Prepare all remote checks before borrowing the SQLite connection.
	for _, sc := range scopes {
		if err := check(ctx, sc); err != nil {
			collect()
			return fmt.Errorf("%w: %w", ErrClaimSQLDenied, err)
		}
	}
	collect()
	database, err := s.sqlDatabaseID()
	if err != nil {
		return err
	}
	var receipts []hostops.SQLReceiptScope
	for _, p := range permits {
		if err = p.BindTransaction(database); err != nil {
			return err
		}
		r := p.Receipt()
		seen := false
		for _, old := range receipts {
			if old == r {
				seen = true
			}
		}
		if !seen {
			receipts = append(receipts, r)
		}
	}
	return s.withPreparedClaimSQL(ctx, scopes, func(context.Context, claimSQLScope) error { return nil }, local, receipts, true, fn)
}

// readTombstoneVersion is release.ReadAppTombstone plus the marker's exact
// version, from one read. (nil, "", nil) when absent.
func readTombstoneVersion(ctx context.Context, rd release.Store, sqid, app string) (*release.AppTombstone, string, error) {
	body, tag, err := rd.Get(ctx, release.AppTombstoneKey(sqid, app))
	if errors.Is(err, os.ErrNotExist) {
		return nil, "", nil
	}
	if err != nil {
		return nil, "", fmt.Errorf("release: read trash marker for %s/%s: %w", sqid, app, err)
	}
	var t release.AppTombstone
	if err := json.Unmarshal(body, &t); err != nil {
		return nil, "", fmt.Errorf("release: malformed trash marker for %s/%s: %w", sqid, app, err)
	}
	if t.Schema != 1 || t.Account != sqid || t.App != app || t.DeletedAt.IsZero() || tag == "" {
		return nil, "", fmt.Errorf("release: trash marker for %s/%s: %w", sqid, app, hostops.ErrIdentity)
	}
	return &t, tag, nil
}

// purgeJournalAbsent refuses while a purge journal exists: an in-flight purge
// stays authoritative over every projection.
func purgeJournalAbsent(ctx context.Context, rd release.Store, sqid, app string) error {
	key, err := hostops.PurgeKey(sqid, app)
	if err != nil {
		return err
	}
	if _, _, err := rd.Get(ctx, key); !errors.Is(err, os.ErrNotExist) {
		if err != nil {
			return err
		}
		return hostops.ErrPending
	}
	return nil
}

// Grant kinds, and the one permit mode each may bind (store_gate.go). The
// mode decides the statement family SQLite lets through: only an effect
// permit inserts or adopts; the terminal projections update or delete
// exactly the incarnation they name.
const (
	grantEffect  = "effect-permit"
	grantRestore = "restore-receipt"
	grantTrash   = "trash-marker"
	grantPurge   = "purge-receipt"
)

var grantModes = map[string]string{
	grantEffect:  claimModeEffect,
	grantRestore: claimModeRestore,
	grantTrash:   claimModeTrash,
	grantPurge:   claimModePurge,
}

func scopeIn(sc claimSQLScope, mode string) claimSQLScope {
	sc.Mode = mode
	return sc
}

// permitGrant is effect-time authority: the live permit EffectRunner.Issue
// put in pctx (an exclusive operation's issued effect, or a live
// registration's) still validates against canonical admission for this
// incarnation. pctx is captured, not taken from each statement: Store
// methods run ctx-less statements, and a request context must not be able to
// widen or narrow the permit set mid-callback.
func permitGrant(pctx context.Context, rd release.Store, sqid string, sc claimSQLScope) sqlGrant {
	var prepared *hostops.PreparedSQLPermit
	return sqlGrant{scope: scopeIn(sc, claimModeEffect), kind: grantEffect, prepared: &prepared,
		check: func(context.Context) error {
			var err error
			prepared, err = hostops.PrepareSQLPermit(pctx, rd, sqid, sc.App, sc.AppID)
			return err
		},
		local: func(ctx context.Context) error {
			if prepared == nil {
				return hostops.ErrPending
			}
			return prepared.Check(ctx)
		}}
}

// restoreReceiptCurrent compares immutable incarnation and exact latest
// completion. Later completed operations may raise the admission epoch. The
// caller separately holds a live permit; this predicate grants no authority.
func restoreReceiptCurrent(a hostops.AppAdmission, r hostops.RestoreMark) bool {
	cur := a.LastRestore
	return cur != nil && *cur == r && r.State == "complete" && r.ID != "" && r.TombstoneETag != "" && r.DeletedAt > 0 &&
		r.AppID != "" && r.AppID == a.AppID && r.Epoch > 0 && r.Epoch <= a.Epoch
}

// restoreReceiptGrant binds the narrow restore SQL mode to a live permit,
// the exact latest receipt and absence of both marker and purge journal.
// The row update also matches the receipt's deleted_at. Capture pctx because
// contextless Store methods still have to validate the operation's permit.
func restoreReceiptGrant(pctx context.Context, rd release.Store, sqid string, sc claimSQLScope, r hostops.RestoreMark) sqlGrant {
	g := permitGrant(pctx, rd, sqid, sc)
	return sqlGrant{scope: scopeIn(sc, claimModeRestore), kind: grantRestore, local: g.local, prepared: g.prepared, check: func(ctx context.Context) error {
		a, err := hostops.ReadAdmission(ctx, rd, sqid, sc.App, sc.AppID)
		if err != nil {
			return err
		}
		if g.check(ctx) != nil || r.AppID != sc.AppID || !restoreReceiptCurrent(a.Value, r) {
			return hostops.ErrPending
		}
		if err := purgeJournalAbsent(ctx, rd, sqid, sc.App); err != nil {
			return err
		}
		tomb, _, err := readTombstoneVersion(ctx, rd, sqid, sc.App)
		if err != nil {
			return err
		}
		if tomb != nil {
			return hostops.ErrPending
		}
		return nil
	}}
}

// tombstoneGrant requires a live permit for this incarnation and the exact
// current marker, with no purge journal. The SQL mode permits only the trash
// columns. The held app operation excludes another console's restore or purge
// through the entire local check and write; a local DB lock alone cannot.
func tombstoneGrant(pctx context.Context, rd release.Store, sqid string, sc claimSQLScope, etag string) sqlGrant {
	g := permitGrant(pctx, rd, sqid, sc)
	return sqlGrant{scope: scopeIn(sc, claimModeTrash), kind: grantTrash, local: g.local, prepared: g.prepared, check: func(ctx context.Context) error {
		_, err := hostops.ReadAdmission(ctx, rd, sqid, sc.App, sc.AppID)
		if err != nil {
			return err
		}
		if g.check(ctx) != nil {
			return hostops.ErrPending
		}
		if err := purgeJournalAbsent(ctx, rd, sqid, sc.App); err != nil {
			return err
		}
		_, seen, err := readTombstoneVersion(ctx, rd, sqid, sc.App)
		if err != nil {
			return err
		}
		if etag == "" || seen != etag {
			return hostops.ErrPending
		}
		return nil
	}}
}

// purgeRetired is positive evidence that appID was retired by purge op:
// last_purge names exactly that incarnation and operation, and either the
// admission is that incarnation's completed retirement, or a successor
// incarnation already replaced it. A successor exists only through
// EnrollNewApp's recreation transition, which requires the completed
// retirement and carries last_purge over, so an unequal app_id counts only
// together with a last_purge naming the retired one.
func purgeRetired(a hostops.AppAdmission, appID, op string) bool {
	lp := a.LastPurge
	if lp == nil || op == "" || appID == "" || lp.ID != op || lp.AppID != appID {
		return false
	}
	if a.AppID == appID {
		return hostops.CompletedAdmission(a)
	}
	return true
}

// purgeReceiptGrant replays a completed purge's SQL cleanup on any console,
// keyed on the retired incarnation, including after the name was recreated:
// the canonical admission (whichever incarnation it now names) carries
// last_purge for scope.AppID and op, and no purge journal exists. Its permit
// is purge mode: DELETE of the retired incarnation's rows, the apps retire and
// claim-metadata UPDATE, and monotone door-key revocation. It cannot create
// an operation, reopen admission or touch a successor's row.
func purgeReceiptGrant(rd release.Store, sqid string, sc claimSQLScope, op string) sqlGrant {
	return sqlGrant{scope: scopeIn(sc, claimModePurge), kind: grantPurge, check: func(ctx context.Context) error {
		a, err := hostops.ReadCanonicalAdmission(ctx, rd, sqid, sc.App)
		if err != nil {
			return err
		}
		if !purgeRetired(a.Value, sc.AppID, op) {
			return hostops.ErrPending
		}
		return purgeJournalAbsent(ctx, rd, sqid, sc.App)
	}}
}