package main

// The store on the mesh's postgres (novox/hq ADR 0290 rule 7): the database the mesh provisions for this module
// (the postgres-database provision), its address in the file DATABASE_URL_FILE names. Its schema is numbered
// migrations in migrations/, applied by the preparation step (`mesh-knowledge prepare`, ADR 0135) and recorded
// in the database itself; a fresh database is given the consolidated schema, schema.sql, at once, and every
// migration is recorded as applied. The serving bundle never changes the schema: it refuses to serve one that is
// behind, and says so.

import (
	"context"
	"crypto/sha256"
	"embed"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io/fs"
	"os"
	"path"
	"regexp"
	"sort"
	"strconv"
	"strings"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"
	"github.com/jackc/pgx/v5/pgxpool"
)

//go:embed migrations/*.sql
var migrationFiles embed.FS

//go:embed schema.sql
var consolidated string

// Migration is one numbered change of the schema.
type Migration struct {
	Number   int
	Name     string
	SQL      string
	Checksum string
}

var migrationFile = regexp.MustCompile(`^(\d{4})-([a-z0-9-]+)\.sql$`)

// Migrations are the migrations this binary carries, in order. A misnamed file, two of one number or an empty
// one is refused: each would make the schema something nobody wrote down.
func Migrations() ([]Migration, error) {
	entries, err := fs.ReadDir(migrationFiles, "migrations")
	if err != nil {
		return nil, err
	}
	var out []Migration
	seen := map[int]string{}
	for _, e := range entries {
		m := migrationFile.FindStringSubmatch(e.Name())
		if m == nil {
			return nil, fmt.Errorf("%s is not a migration's name (NNNN-what-it-does.sql)", e.Name())
		}
		n, _ := strconv.Atoi(m[1])
		if other, twice := seen[n]; twice {
			return nil, fmt.Errorf("two migrations are numbered %04d: %s and %s", n, other, e.Name())
		}
		seen[n] = e.Name()
		body, err := fs.ReadFile(migrationFiles, path.Join("migrations", e.Name()))
		if err != nil {
			return nil, err
		}
		if strings.TrimSpace(string(body)) == "" {
			return nil, fmt.Errorf("%s is empty", e.Name())
		}
		sum := sha256.Sum256(body)
		out = append(out, Migration{Number: n, Name: m[2], SQL: string(body), Checksum: hex.EncodeToString(sum[:])})
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Number < out[j].Number })
	for i, m := range out {
		if m.Number != i+1 {
			return nil, fmt.Errorf("the migrations are not numbered 0001 on without a gap: %04d is the %d. one", m.Number, i+1)
		}
	}
	return out, nil
}

// ledger is the record of what ran, kept in the database it describes.
const ledger = `create table if not exists migration (
    number   integer     primary key,
    name     text        not null,
    checksum text        not null,
    applied  timestamptz not null default now())`

// lockKey is the advisory lock every migration run takes, so two preparations never interleave.
const lockKey int64 = 4_250_290_425

// PgStore is the Store on postgres.
type PgStore struct{ pool *pgxpool.Pool }

// PgStoreFromEnv opens the store the mesh provisioned, its address in the file DATABASE_URL_FILE names.
func PgStoreFromEnv(ctx context.Context) (*PgStore, error) {
	file := os.Getenv("DATABASE_URL_FILE")
	if file == "" {
		return nil, errors.New("DATABASE_URL_FILE is not set: the knowledge base's database is a requirement the mesh resolves")
	}
	raw, err := os.ReadFile(file)
	if err != nil {
		return nil, fmt.Errorf("the database's address cannot be read: %w", err)
	}
	return OpenPgStore(ctx, strings.TrimSpace(string(raw)))
}

// OpenPgStore opens a store at an address.
func OpenPgStore(ctx context.Context, url string) (*PgStore, error) {
	pool, err := pgxpool.New(ctx, url)
	if err != nil {
		return nil, err
	}
	return &PgStore{pool: pool}, nil
}

// Close closes the store.
func (s *PgStore) Close() { s.pool.Close() }

// Prepare brings the schema to what this version needs, and says what it did: on a database that holds nothing
// of it, the consolidated schema with every migration recorded; otherwise every migration not yet applied, each
// in its own transaction with its row in the ledger. A migration that ran and whose file changed since, or one
// numbered below the highest applied that never ran, stops it: the database's contents are then not known.
func (s *PgStore) Prepare(ctx context.Context) ([]string, error) {
	migrations, err := Migrations()
	if err != nil {
		return nil, err
	}
	conn, err := s.pool.Acquire(ctx)
	if err != nil {
		return nil, err
	}
	defer conn.Release()
	if _, err := conn.Exec(ctx, `select pg_advisory_lock($1)`, lockKey); err != nil {
		return nil, fmt.Errorf("the migration lock cannot be taken: %w", err)
	}
	defer func() { _, _ = conn.Exec(context.WithoutCancel(ctx), `select pg_advisory_unlock($1)`, lockKey) }()
	if _, err := conn.Exec(ctx, ledger); err != nil {
		return nil, err
	}
	applied, err := appliedMigrations(ctx, conn.Conn())
	if err != nil {
		return nil, err
	}
	var said []string
	if len(applied) == 0 {
		var exists bool
		if err := conn.QueryRow(ctx, `select to_regclass('record') is not null`).Scan(&exists); err != nil {
			return nil, err
		}
		if exists {
			return nil, errors.New("the database holds the knowledge base's tables and no record of a migration: its contents are not known, so nothing is changed")
		}
		tx, err := conn.Begin(ctx)
		if err != nil {
			return nil, err
		}
		defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
		if _, err := tx.Exec(ctx, consolidated); err != nil {
			return nil, fmt.Errorf("the consolidated schema failed, and nothing of it was kept: %w", err)
		}
		for _, m := range migrations {
			if _, err := tx.Exec(ctx, `insert into migration (number, name, checksum) values ($1, $2, $3)`, m.Number, m.Name, m.Checksum); err != nil {
				return nil, err
			}
		}
		if err := tx.Commit(ctx); err != nil {
			return nil, err
		}
		return []string{fmt.Sprintf("a fresh database: the consolidated schema, through migration %04d", len(migrations))}, nil
	}
	highest := 0
	var problems []string
	byNumber := map[int]appliedMigration{}
	for _, a := range applied {
		byNumber[a.number] = a
		highest = max(highest, a.number)
	}
	var pending []Migration
	for _, m := range migrations {
		a, ran := byNumber[m.Number]
		switch {
		case ran && a.checksum != m.Checksum:
			problems = append(problems, fmt.Sprintf("migration %04d-%s ran, and its file has changed since: write a new one", m.Number, m.Name))
		case !ran && m.Number < highest:
			problems = append(problems, fmt.Sprintf("migration %04d-%s never ran, and %04d did: renumber it above %04d", m.Number, m.Name, highest, highest))
		case !ran:
			pending = append(pending, m)
		}
	}
	if len(problems) > 0 {
		return nil, errors.New("this database and these migrations disagree: " + strings.Join(problems, "; "))
	}
	for _, m := range pending {
		err := func() error {
			tx, err := conn.Begin(ctx)
			if err != nil {
				return err
			}
			defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
			if _, err := tx.Exec(ctx, m.SQL); err != nil {
				return fmt.Errorf("migration %04d-%s failed, and nothing it did was kept: %w", m.Number, m.Name, err)
			}
			if _, err := tx.Exec(ctx, `insert into migration (number, name, checksum) values ($1, $2, $3)`, m.Number, m.Name, m.Checksum); err != nil {
				return err
			}
			return tx.Commit(ctx)
		}()
		if err != nil {
			return said, err
		}
		said = append(said, fmt.Sprintf("migration %04d-%s applied", m.Number, m.Name))
	}
	if len(said) == 0 {
		said = []string{fmt.Sprintf("the schema is at migration %04d, as this version needs", len(migrations))}
	}
	return said, nil
}

type appliedMigration struct {
	number   int
	checksum string
}

func appliedMigrations(ctx context.Context, conn *pgx.Conn) ([]appliedMigration, error) {
	rows, err := conn.Query(ctx, `select number, checksum from migration order by number`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []appliedMigration
	for rows.Next() {
		var a appliedMigration
		if err := rows.Scan(&a.number, &a.checksum); err != nil {
			return nil, err
		}
		out = append(out, a)
	}
	return out, rows.Err()
}

// Ready says why the store cannot serve this version, or nil: the schema must be at the migrations it carries.
func (s *PgStore) Ready(ctx context.Context) error {
	migrations, err := Migrations()
	if err != nil {
		return err
	}
	var at int
	err = s.pool.QueryRow(ctx, `select coalesce(max(number), 0) from migration`).Scan(&at)
	var pgErr *pgconn.PgError
	if errors.As(err, &pgErr) && pgErr.Code == "42P01" {
		at = 0
	} else if err != nil {
		return fmt.Errorf("the database cannot be reached: %w", err)
	}
	if at != len(migrations) {
		return fmt.Errorf("the database's schema is at migration %04d and this version needs %04d: the preparation step "+
			"(mesh-knowledge prepare) has not run", at, len(migrations))
	}
	return nil
}

const entryColumns = `r.kind, r.number, r.current_version, coalesce(r.pending_version, 0), r.revision,
  (select count(*) from version v where v.kind = r.kind and v.number = r.number),
  (select body from version v where v.kind = r.kind and v.number = r.number and v.n = r.current_version),
  (select body from version v where v.kind = r.kind and v.number = r.number and v.n = r.pending_version)`

func scanEntry(row pgx.Row) (*Entry, error) {
	var e Entry
	var kind string
	var number int
	var current, pending []byte
	if err := row.Scan(&kind, &number, &e.CurrentVersion, &e.PendingVersion, &e.Revision, &e.Versions, &current, &pending); err != nil {
		return nil, err
	}
	e.Current = &Record{}
	if err := json.Unmarshal(current, e.Current); err != nil {
		return nil, fmt.Errorf("%s's version %d cannot be read: %w", Key{kind, number}, e.CurrentVersion, err)
	}
	if len(pending) > 0 {
		e.Pending = &Record{}
		if err := json.Unmarshal(pending, e.Pending); err != nil {
			return nil, fmt.Errorf("%s's version %d cannot be read: %w", Key{kind, number}, e.PendingVersion, err)
		}
	}
	return &e, nil
}

func (s *PgStore) Get(ctx context.Context, k Key) (*Entry, error) {
	e, err := scanEntry(s.pool.QueryRow(ctx, `select `+entryColumns+` from record r where r.kind = $1 and r.number = $2`, k.Kind, k.Number))
	if errors.Is(err, pgx.ErrNoRows) {
		return nil, nil
	}
	return e, err
}

func (s *PgStore) All(ctx context.Context) ([]*Entry, error) {
	rows, err := s.pool.Query(ctx, `select `+entryColumns+` from record r order by r.kind, r.number`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []*Entry
	for rows.Next() {
		e, err := scanEntry(rows)
		if err != nil {
			return nil, err
		}
		out = append(out, e)
	}
	return out, rows.Err()
}

func sectionsText(r *Record) string {
	var b strings.Builder
	for _, s := range r.Sections {
		b.WriteString(s.Text)
		b.WriteString("\n")
	}
	return b.String()
}

// served writes the rows of the version a record serves: its status, topic and search vector, its clauses and
// its links.
func served(ctx context.Context, tx pgx.Tx, r *Record) error {
	if _, err := tx.Exec(ctx, `update record set status = $3, topic = $4,
	    search = setweight(to_tsvector('english', $5), 'A') || setweight(to_tsvector('english', $4), 'B') ||
	             setweight(to_tsvector('english', $6), 'D')
	  where kind = $1 and number = $2`, r.Kind, r.Number, r.Status, r.Topic, r.Title, sectionsText(r)); err != nil {
		return err
	}
	if _, err := tx.Exec(ctx, `delete from clause where kind = $1 and number = $2`, r.Kind, r.Number); err != nil {
		return err
	}
	for _, c := range r.Clauses {
		if _, err := tx.Exec(ctx, `insert into clause (kind, number, n, text, status, source, search)
		  values ($1, $2, $3, $4, $5, $6, setweight(to_tsvector('english', $4), 'A'))`,
			r.Kind, r.Number, c.N, c.Text, c.Status, c.Source); err != nil {
			return err
		}
	}
	if _, err := tx.Exec(ctx, `delete from link where from_kind = $1 and from_number = $2`, r.Kind, r.Number); err != nil {
		return err
	}
	for _, l := range r.Links {
		body, _ := json.Marshal(l)
		if _, err := tx.Exec(ctx, `insert into link (from_kind, from_number, from_clause, kind, to_kind, to_number, to_clause, to_path, body)
		  values ($1, $2, $3, $4, $5, $6, $7, $8, $9)`, r.Kind, r.Number, l.From, l.Kind, l.To.Kind, l.To.Number, l.To.Clause, l.To.Path, body); err != nil {
			return err
		}
	}
	return nil
}

func insertVersion(ctx context.Context, tx pgx.Tx, v Version) error {
	body, err := json.Marshal(v.Record)
	if err != nil {
		return err
	}
	_, err = tx.Exec(ctx, `insert into version (kind, number, n, by, at, why, insight, body) values ($1, $2, $3, $4, $5, $6, $7, $8)`,
		v.Record.Kind, v.Record.Number, v.N, v.By, v.At, v.Why, v.Insight, body)
	return err
}

func (s *PgStore) Create(ctx context.Context, v Version) error {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
	r := v.Record
	tag, err := tx.Exec(ctx, `insert into record (kind, number, status, topic, current_version) values ($1, $2, $3, $4, 1)
	  on conflict do nothing`, r.Kind, r.Number, r.Status, r.Topic)
	if err != nil {
		return err
	}
	if tag.RowsAffected() == 0 {
		return ErrExists
	}
	v.N = 1
	if err := insertVersion(ctx, tx, v); err != nil {
		return err
	}
	if err := served(ctx, tx, r); err != nil {
		return err
	}
	return tx.Commit(ctx)
}

func (s *PgStore) Save(ctx context.Context, v Version, current, clearPending bool, revision int64) error {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
	r := v.Record
	var held int64
	var versions int
	err = tx.QueryRow(ctx, `select revision, (select count(*) from version v where v.kind = r.kind and v.number = r.number)
	  from record r where kind = $1 and number = $2 for update`, r.Kind, r.Number).Scan(&held, &versions)
	if errors.Is(err, pgx.ErrNoRows) {
		return fmt.Errorf("%s is not held", r.Key())
	}
	if err != nil {
		return err
	}
	if held != revision {
		return ErrChanged
	}
	v.N = versions + 1
	if err := insertVersion(ctx, tx, v); err != nil {
		return err
	}
	if current {
		if _, err := tx.Exec(ctx, `update record set current_version = $3 where kind = $1 and number = $2`, r.Kind, r.Number, v.N); err != nil {
			return err
		}
		if err := served(ctx, tx, r); err != nil {
			return err
		}
	} else {
		if _, err := tx.Exec(ctx, `update record set pending_version = $3 where kind = $1 and number = $2`, r.Kind, r.Number, v.N); err != nil {
			return err
		}
	}
	if clearPending {
		if _, err := tx.Exec(ctx, `update record set pending_version = null where kind = $1 and number = $2`, r.Kind, r.Number); err != nil {
			return err
		}
	}
	if _, err := tx.Exec(ctx, `update record set revision = revision + 1 where kind = $1 and number = $2`, r.Kind, r.Number); err != nil {
		return err
	}
	return tx.Commit(ctx)
}

func (s *PgStore) Versions(ctx context.Context, k Key) ([]Version, error) {
	rows, err := s.pool.Query(ctx, `select n, by, at, why, insight, body from version where kind = $1 and number = $2 order by n`, k.Kind, k.Number)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []Version
	for rows.Next() {
		var v Version
		var body []byte
		if err := rows.Scan(&v.N, &v.By, &v.At, &v.Why, &v.Insight, &body); err != nil {
			return nil, err
		}
		v.Record = &Record{}
		if err := json.Unmarshal(body, v.Record); err != nil {
			return nil, err
		}
		out = append(out, v)
	}
	return out, rows.Err()
}

// Search ranks by the database's full-text search: any of the question's words, a clause weighing more than a
// record's sections, a title most.
func (s *PgStore) Search(ctx context.Context, question string, limit int) ([]Hit, error) {
	words := questionWords(question)
	if len(words) == 0 {
		return nil, nil
	}
	if limit <= 0 {
		limit = 200
	}
	rows, err := s.pool.Query(ctx, `with q as (select replace(plainto_tsquery('english', $1)::text, ' & ', ' | ')::tsquery as q)
	  select kind, number, clause, rank from (
	    select c.kind, c.number, c.n as clause, ts_rank_cd(c.search, q.q) as rank
	      from clause c join record r on r.kind = c.kind and r.number = c.number, q
	      where c.search @@ q.q and r.status not in ('reserved', 'withdrawn')
	    union all
	    select r.kind, r.number, 0, ts_rank_cd(r.search, q.q)
	      from record r, q where r.search @@ q.q and r.status not in ('reserved', 'withdrawn')
	  ) hits order by rank desc, number desc, clause limit $2`, strings.Join(words, " "), limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []Hit
	for rows.Next() {
		var h Hit
		if err := rows.Scan(&h.Key.Kind, &h.Key.Number, &h.Clause, &h.Rank); err != nil {
			return nil, err
		}
		out = append(out, h)
	}
	return out, rows.Err()
}

func (s *PgStore) Inbound(ctx context.Context, k Key) ([]Inbound, error) {
	rows, err := s.pool.Query(ctx, `select from_kind, from_number, body from link where to_kind = $1 and to_number = $2
	  order by from_number, kind`, k.Kind, k.Number)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []Inbound
	for rows.Next() {
		var in Inbound
		var body []byte
		if err := rows.Scan(&in.From.Kind, &in.From.Number, &body); err != nil {
			return nil, err
		}
		if err := json.Unmarshal(body, &in.Link); err != nil {
			return nil, err
		}
		out = append(out, in)
	}
	return out, rows.Err()
}

func (s *PgStore) SetFindings(ctx context.Context, f []Finding) error {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
	if _, err := tx.Exec(ctx, `delete from finding`); err != nil {
		return err
	}
	for _, x := range f {
		if _, err := tx.Exec(ctx, `insert into finding (rule, records, says, at) values ($1, $2, $3, $4)`, x.Rule, x.Records, x.Says, x.At); err != nil {
			return err
		}
	}
	return tx.Commit(ctx)
}

func (s *PgStore) Findings(ctx context.Context) ([]Finding, error) {
	rows, err := s.pool.Query(ctx, `select rule, records, says, at from finding order by rule, says`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []Finding
	for rows.Next() {
		var f Finding
		if err := rows.Scan(&f.Rule, &f.Records, &f.Says, &f.At); err != nil {
			return nil, err
		}
		out = append(out, f)
	}
	return out, rows.Err()
}

func (s *PgStore) ToldOf(ctx context.Context, prefix string) ([]string, error) {
	// left() and not like: a prefix may hold an underscore, which like would take for any character.
	rows, err := s.pool.Query(ctx, `select topic from told where left(topic, length($1)) = $1 order by topic`, prefix)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []string
	for rows.Next() {
		var k string
		if err := rows.Scan(&k); err != nil {
			return nil, err
		}
		out = append(out, k)
	}
	return out, rows.Err()
}

func (s *PgStore) Untold(ctx context.Context, key string) error {
	_, err := s.pool.Exec(ctx, `delete from told where topic = $1`, key)
	return err
}

func (s *PgStore) Told(ctx context.Context, topic string) (bool, error) {
	tag, err := s.pool.Exec(ctx, `insert into told (topic) values ($1) on conflict do nothing`, topic)
	if err != nil {
		return false, err
	}
	return tag.RowsAffected() == 1, nil
}