package console

import (
	"context"
	"crypto/rand"
	"database/sql"
	"database/sql/driver"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/carlosframework/platform/internal/hostops"
	"log"
	"time"
)

// These tables exist only in the claim-on console's existing durable DB.
// The lexer blocks all ordinary handles from naming either table.
func initSQLReceipts(db *sql.DB, instance string) error {
	_, err := db.Exec(`CREATE TABLE IF NOT EXISTS claim_sql_identity (id TEXT PRIMARY KEY, instance TEXT NOT NULL DEFAULT '');
 CREATE TABLE IF NOT EXISTS claim_sql_receipts (effect TEXT PRIMARY KEY, scope TEXT NOT NULL, state TEXT NOT NULL CHECK(state IN ('committed','fenced')));`)
	if err != nil {
		return err
	}
	var hasInstance int
	if err = db.QueryRow(`SELECT count(*) FROM pragma_table_info('claim_sql_identity') WHERE name = 'instance'`).Scan(&hasInstance); err != nil {
		return err
	}
	if hasInstance == 0 {
		if _, err = db.Exec(`ALTER TABLE claim_sql_identity ADD COLUMN instance TEXT NOT NULL DEFAULT ''`); err != nil {
			return err
		}
	}
	var nonce [24]byte
	if _, err = rand.Read(nonce[:]); err != nil {
		return err
	}
	if _, err = db.Exec(`INSERT INTO claim_sql_identity(id, instance) SELECT ?, ? WHERE NOT EXISTS(SELECT 1 FROM claim_sql_identity)`, hex.EncodeToString(nonce[:]), instance); err != nil {
		return err
	}
	// A row from before the instance was bound takes this instance. That is
	// safe because the instance is part of the database ID: effects bound to
	// the old ID stay unsettled here, and a clone adopting the same row gets
	// an ID of its own.
	if _, err = db.Exec(`UPDATE claim_sql_identity SET instance = ? WHERE instance = ''`, instance); err != nil {
		return err
	}
	var rows int
	var stored string
	if err = db.QueryRow(`SELECT count(*), COALESCE(MIN(instance), '') FROM claim_sql_identity`).Scan(&rows, &stored); err != nil {
		return err
	}
	if rows != 1 {
		return fmt.Errorf("claim SQL receipt database has %d identity rows: %w", rows, hostops.ErrIdentity)
	}
	if stored == instance {
		return nil
	}
	// Restored onto another instance (review I2-2): adopt it under a fresh
	// nonce, so this copy has a database ID no other copy has had, even one
	// moved back to an instance it lived on before. Copied receipts still
	// cannot settle the original's effects, because every settlement compares
	// the effect's database ID first; those effects stay unsettled and are
	// alerted past sql_recover_after.
	if _, err = rand.Read(nonce[:]); err != nil {
		return err
	}
	if _, err = db.Exec(`UPDATE claim_sql_identity SET id = ?, instance = ?`, hex.EncodeToString(nonce[:]), instance); err != nil {
		return err
	}
	log.Printf("console: claim SQL receipt database moved from instance %s to %s; it has a new database ID, and effects bound to the old one stay unsettled", stored, instance)
	return nil
}

// databaseID is the receipt database's full identity: the stored nonce, the
// instance verified at open, and the physical file, re-read on every call so
// a database replaced under the running process is noticed.
func (g *claimGate) databaseID(nonce string) (string, error) {
	physical, err := sqlPhysicalIdentity(g.path)
	if err != nil {
		return "", err
	}
	if physical != g.physical || g.instance == "" {
		return "", hostops.ErrIdentity
	}
	return nonce + ":" + g.instance + ":" + physical, nil
}
func cloneSQLReceiptTables(in map[string]claimSQLTable) map[string]claimSQLTable {
	out := make(map[string]claimSQLTable, len(in)+2)
	for k, v := range in {
		out[k] = v
	}
	out["claim_sql_identity"] = claimSQLTable{NonApp: "private SQL receipt database identity; ordinary SQL handles refuse access"}
	out["claim_sql_receipts"] = claimSQLTable{NonApp: "private effect receipts and fences; ordinary SQL handles refuse access"}
	return out
}
func (s *Store) sqlDatabaseID() (string, error) {
	if s == nil || s.gate == nil || s.rawBoot == nil {
		return "", hostops.ErrContract
	}
	var id string
	err := s.rawBoot.QueryRow(`SELECT id FROM claim_sql_identity`).Scan(&id)
	if err != nil {
		return "", err
	}
	return s.gate.databaseID(id)
}

// FenceSQL serializes with the original transaction on SQLite's writer lock.
// Inserting the durable fence even when a receipt is absent closes the
// issued-before-BEGIN crash window. A delayed BEGIN can only refuse it.
func (s *Store) FenceSQL(ctx context.Context, sc hostops.SQLReceiptScope) error {
	_, err := s.FenceSQLOutcome(ctx, sc)
	return err
}

// FenceSQLOutcome is FenceSQL that also reports whether the original
// transaction's committed receipt was found, so recovery records what it
// proved rather than calling every outcome a fence.
func (s *Store) FenceSQLOutcome(ctx context.Context, sc hostops.SQLReceiptScope) (committed bool, err error) {
	if s.gate == nil || sc.Effect == "" || sc.Operation == "" || sc.AppID == "" || sc.Epoch == 0 {
		return false, hostops.ErrIdentity
	}
	id, err := s.sqlDatabaseID()
	if err != nil {
		return false, err
	}
	if id != sc.Database {
		return false, hostops.ErrIdentity
	}
	body, err := json.Marshal(sc)
	if err != nil {
		return false, err
	}
	conn, err := s.rawBoot.Conn(ctx)
	if err != nil {
		return false, err
	}
	defer conn.Close()
	if _, err = conn.ExecContext(ctx, "BEGIN IMMEDIATE"); err != nil {
		return false, err
	}
	defer conn.ExecContext(context.Background(), "ROLLBACK")
	var have, state string
	err = conn.QueryRowContext(ctx, `SELECT scope, state FROM claim_sql_receipts WHERE effect=?`, sc.Effect).Scan(&have, &state)
	if err != nil && !errors.Is(err, sql.ErrNoRows) {
		return false, err
	}
	if err == nil && have != string(body) {
		return false, hostops.ErrIdentity
	}
	if _, err = conn.ExecContext(ctx, `INSERT INTO claim_sql_receipts(effect,scope,state) VALUES(?,?,'fenced') ON CONFLICT(effect) DO NOTHING`, sc.Effect, string(body)); err != nil {
		return false, err
	}
	if _, err = conn.ExecContext(ctx, "COMMIT"); err != nil {
		return false, err
	}
	return state == "committed", nil
}

type claimSavepoint struct{ c *claimConn }

func (t *claimSavepoint) Commit() error {
	return t.c.innerExec(context.Background(), "RELEASE claim_batch_statement")
}
func (t *claimSavepoint) Rollback() error {
	if err := t.c.innerExec(context.Background(), "ROLLBACK TO claim_batch_statement"); err != nil {
		return err
	}
	return t.c.innerExec(context.Background(), "RELEASE claim_batch_statement")
}
func (c *claimConn) beginBatch(ctx context.Context, b *claimBinding) error {
	// IMMEDIATE is required: a paused deferred read transaction cannot upgrade
	// after recovery inserted a fence using another connection.
	if err := c.innerExec(ctx, "BEGIN IMMEDIATE"); err != nil {
		return err
	}
	b.outer = &claimBatchTx{c: c}
	if len(b.receipts) == 0 {
		return nil
	}
	for _, sc := range b.receipts {
		rows, err := c.inner.QueryContext(ctx, `SELECT id FROM claim_sql_identity`, nil)
		if err != nil {
			return err
		}
		vals := make([]driver.Value, 1)
		err = rows.Next(vals)
		rows.Close()
		if err != nil {
			return err
		}
		nonce, ok := vals[0].(string)
		if !ok {
			return hostops.ErrIdentity
		}
		id, err := c.g.databaseID(nonce)
		if err != nil {
			return err
		}
		if id != sc.Database {
			return hostops.ErrIdentity
		}
		// Any prior terminal row makes replay impossible, including a recovery
		// fence written before the first SQL call reached this connection.
		rows, err = c.inner.QueryContext(ctx, `SELECT count(*) FROM claim_sql_receipts WHERE effect=?`, []driver.NamedValue{{Ordinal: 1, Value: sc.Effect}})
		if err != nil {
			return err
		}
		vals = make([]driver.Value, 1)
		err = rows.Next(vals)
		rows.Close()
		n, _ := vals[0].(int64)
		if err != nil {
			return err
		}
		if n != 0 {
			return hostops.ErrPending
		}
	}
	return nil
}

type claimBatchTx struct{ c *claimConn }

func (t *claimBatchTx) Commit() error   { return t.c.innerExec(context.Background(), "COMMIT") }
func (t *claimBatchTx) Rollback() error { return t.c.innerExec(context.Background(), "ROLLBACK") }
func (c *claimConn) endBatch(b *claimBinding, commit bool) error {
	if b.outer == nil {
		return hostops.ErrContract
	}
	// The caller sealed every public handle; this runs under the driver lock,
	// so nothing else is in flight. Finish what the handles left behind.
	b.sealed.Store(true)
	c.closeRows()
	if c.tx != nil {
		c.tx.abort()
		if commit {
			// A transaction left open means the requested commit cannot be
			// honoured: roll back and say so, never report success.
			err := fmt.Errorf("%w: transaction still open at batch commit", ErrClaimSQLScope)
			if rerr := b.outer.Rollback(); rerr != nil {
				c.bad = true
				err = errors.Join(err, rerr)
			}
			b.outer = nil
			return err
		}
	}
	if commit {
		err := b.lostErr()
		if err == nil {
			err = b.authorizeCommit(b.ctx, c)
		}
		if err != nil {
			if rerr := b.outer.Rollback(); rerr != nil {
				c.bad = true
				err = errors.Join(err, rerr)
			}
			b.outer = nil
			return err
		}
	}
	if commit {
		for _, receipt := range b.receipts {
			body, err := json.Marshal(receipt)
			if err != nil {
				return err
			}
			if err = c.innerExec(b.ctx, `INSERT INTO claim_sql_receipts(effect,scope,state) VALUES(?,?,'committed')`, receipt.Effect, string(body)); err != nil {
				return err
			}
		}
	}
	var err error
	if commit {
		err = b.outer.Commit()
	} else {
		err = b.outer.Rollback()
	}
	b.outer = nil
	if err != nil {
		c.bad = true
	}
	return err
}

// A returned SQL error is not necessarily ambiguous: after handles joined,
// the same positive local fence decides it immediately. Keep the callback's
// error for its caller while settling authority only after this proof.
func (s *Store) finishPreparedSQL(ctx context.Context, p *hostops.PreparedSQLPermit, callErr error) error {
	if callErr != nil {
		proofCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 20*time.Second)
		defer cancel()
		if err := s.FenceSQL(proofCtx, p.Receipt()); err != nil {
			return errors.Join(err, p.FinishTransaction(callErr))
		}
	}
	return p.FinishTransaction(nil)
}