package main

// What retired means in postgres (novox/hq ADR 0570 clause 5, ADR 0114 §2 and §9; ADR 0230 for the delete).
//
// **Retired is the login removed, the database kept under a keeper role, and the keeper marked.** Retiring happens
// only on the operator's yes to a retiring provision (the SDK's moved path, Gone). What the login owns, its database
// first, is handed to its keeper, a role of its own that cannot log in (`<login>~retired`, KeeperOf), its open
// sessions are ended, and the login is dropped: nothing can connect as it any more, and no credential of it is left.
// The database, its grants to others and every byte in it stay: the backup contribution dumps every database and a
// retired one is still worth backing up. The mark is the keeper's comment, a JSON object the mesh owns:
//
//	{"mesh":"consumer","node":"ace"}                                                       an active login
//	{"mesh":"consumer","node":"ace","retired":"2026-10-06T…Z","why":"…","login":"…"}       a keeper
//
// Asked for again, the ordinary create makes the login again and its keeper hands everything back (Reclaim) and
// goes. Deleted — only through `cleanup delete` — the database is dropped, then the keeper.
//
// **A login retired before this record** was only locked (`NOLOGIN`, its own role marked retired); it is listed and
// deleted as before, and create sets its LOGIN again.
//
// **What the mesh made is recognised by its mark, or by its shape before the mark existed**: a role
// valid until 'infinity' (only the mesh's role statement says that) that owns a database of its own
// name. A role of any other shape is somebody else's and is never listed, retired or deleted. A database
// renamed aside — `<name>_deleted_<yyyymmdd>`, by postgres_retire_database or by hand — is listed as
// retired too, since that date, so a person sees it and can delete it.

import (
	"context"
	"encoding/json"
	"fmt"
	"git.novox.be/novox/mesh-sdk/go/provider/harness"
	"regexp"
	"strconv"
	"time"
)

// KindSetAside is a database renamed aside, listed for deletion beside the retired consumers.
const KindSetAside = "set-aside-database"

// roleMark is the comment the mesh keeps on a consumer's role.
type roleMark struct {
	Mesh    string `json:"mesh"`
	Node    string `json:"node,omitempty"`
	Retired string `json:"retired,omitempty"`
	Why     string `json:"why,omitempty"`
	// Login is, on a keeper, the login it keeps for.
	Login string `json:"login,omitempty"`
}

// KeeperOf is the role a retired login's database and objects are handed to: the login's name with `~retired`.
func KeeperOf(login string) (string, error) {
	k := login + "~retired"
	if len(k) > 63 {
		return "", fmt.Errorf("%s is too long a login for a keeper role (%s is over postgres's 63 bytes)", login, k)
	}
	return k, nil
}

func readMark(comment string) (roleMark, bool) {
	var m roleMark
	if comment == "" || json.Unmarshal([]byte(comment), &m) != nil || m.Mesh != harness.KindConsumer {
		return roleMark{}, false
	}
	return m, true
}

// MarkStatement is the comment that marks a role as the mesh's consumer, active or retired.
func MarkStatement(role string, m roleMark) string {
	m.Mesh = harness.KindConsumer
	b, _ := json.Marshal(m)
	return fmt.Sprintf("COMMENT ON ROLE %s IS %s", Ident(role), Literal(string(b)))
}

// MarkActive writes the active mark — after create, which has already set LOGIN.
func (c *Client) MarkActive(ctx context.Context, role, node string) error {
	_, err := c.Query(ctx, "", MarkStatement(role, roleMark{Node: node}))
	return err
}

// RetireRole removes a login on the operator's yes (ADR 0570 clause 5): what it owns is handed to its keeper, marked
// with when and why, its sessions end and the login is dropped. The database and its data are kept. A login that does
// not exist has nothing to retire.
func (c *Client) RetireRole(ctx context.Context, role, why string, at time.Time) error {
	r, err := c.Query(ctx, "", "SELECT coalesce(shobj_description(oid, 'pg_authid'), '') FROM pg_roles WHERE rolname = "+
		Literal(role))
	if err != nil {
		return err
	}
	if len(r.Rows) == 0 {
		return nil
	}
	prev, _ := readMark(cell(r.Rows[0], 0))
	keeper, err := KeeperOf(role)
	if err != nil {
		return err
	}
	has, err := c.exists(ctx, "SELECT 1 FROM pg_roles WHERE rolname = "+Literal(keeper))
	if err != nil {
		return err
	}
	if !has {
		if _, err := c.Query(ctx, "", fmt.Sprintf("CREATE ROLE %s NOLOGIN", Ident(keeper))); err != nil {
			return err
		}
	}
	if _, err := c.Query(ctx, "", MarkStatement(keeper, roleMark{Node: prev.Node, Retired: at.UTC().Format(time.RFC3339),
		Why: why, Login: role})); err != nil {
		return err
	}
	// No new session, and none left open.
	if err := c.LockRole(ctx, role); err != nil {
		return err
	}
	// In its own database, where its objects are; the database itself is a shared object, handed with them.
	in := ""
	if own, err := c.exists(ctx, "SELECT 1 FROM pg_database WHERE datname = "+Literal(role)); err != nil {
		return err
	} else if own {
		in = role
	}
	for _, sql := range []string{
		fmt.Sprintf("REASSIGN OWNED BY %s TO %s", Ident(role), Ident(keeper)),
		fmt.Sprintf("DROP OWNED BY %s", Ident(role)),
	} {
		if _, err := c.Query(ctx, in, sql); err != nil {
			return err
		}
	}
	_, err = c.Query(ctx, "", fmt.Sprintf("DROP ROLE %s", Ident(role)))
	return err
}

// Reclaim gives a login made again what its keeper kept (ADR 0570 clause 5: a provision needed again): the keeper's
// database and objects are handed back, and the keeper goes. A login with no keeper has nothing to reclaim.
func (c *Client) Reclaim(ctx context.Context, database, role string) error {
	keeper, err := KeeperOf(role)
	if err != nil {
		return err
	}
	has, err := c.exists(ctx, "SELECT 1 FROM pg_roles WHERE rolname = "+Literal(keeper))
	if err != nil || !has {
		return err
	}
	for _, sql := range []string{
		fmt.Sprintf("REASSIGN OWNED BY %s TO %s", Ident(keeper), Ident(role)),
		fmt.Sprintf("DROP OWNED BY %s", Ident(keeper)),
	} {
		if _, err := c.Query(ctx, database, sql); err != nil {
			return err
		}
	}
	_, err = c.Query(ctx, "", fmt.Sprintf("DROP ROLE %s", Ident(keeper)))
	return err
}

var setAside = regexp.MustCompile(`_deleted_(\d{8})$`)

// InventoryStatement lists every role that may be the mesh's, with its login, mark, shape and the size
// of its own database.
const InventoryStatement = `SELECT r.rolname, r.rolcanlogin, coalesce(shobj_description(r.oid, 'pg_authid'), '') AS mark,
  coalesce(r.rolvaliduntil = 'infinity', false) AS mesh_shaped,
  (SELECT pg_database_size(d.datname) FROM pg_database d WHERE d.datdba = r.oid
    AND (d.datname = r.rolname OR d.datname || '~retired' = r.rolname)) AS size
FROM pg_roles r
WHERE r.rolname !~ '^pg_' AND NOT r.rolsuper AND r.rolname <> ` + "'" + Reader + "'" + `
ORDER BY r.rolname`

// SetAsideStatement lists the databases renamed aside.
const SetAsideStatement = `SELECT datname, pg_database_size(datname) FROM pg_database WHERE datname ~ '_deleted_[0-9]{8}$' ORDER BY datname`

// Inventory is what this server holds that the mesh made.
func (c *Client) Inventory(ctx context.Context) (harness.Inventory, error) {
	inv := harness.Inventory{Active: []string{}, Retired: []harness.Retired{}, Sizes: map[string]int64{}}
	r, err := c.Query(ctx, "", InventoryStatement)
	if err != nil {
		return inv, err
	}
	for _, row := range r.Rows {
		name, login, comment, shaped, size := cell(row, 0), cell(row, 1) == "t", cell(row, 2), cell(row, 3) == "t", cell(row, 4)
		m, marked := readMark(comment)
		if !marked && !(shaped && size != "") {
			continue // not the mesh's
		}
		if login && m.Retired == "" {
			inv.Active = append(inv.Active, name)
			if s := sizeOf(size); s >= 0 {
				inv.Sizes[name] = s
			}
			continue
		}
		at, _ := time.Parse(time.RFC3339, m.Retired)
		if m.Login != "" {
			name = m.Login // a keeper: listed under the login it keeps for
		}
		inv.Retired = append(inv.Retired, harness.Retired{Consumer: name, Node: m.Node, RetiredAt: at, Why: m.Why,
			SizeBytes: sizeOf(size), Kind: harness.KindConsumer})
	}
	r, err = c.Query(ctx, "", SetAsideStatement)
	if err != nil {
		return inv, err
	}
	for _, row := range r.Rows {
		name := cell(row, 0)
		m := setAside.FindStringSubmatch(name)
		if m == nil {
			continue
		}
		at, _ := time.Parse("20060102", m[1])
		inv.Retired = append(inv.Retired, harness.Retired{Consumer: name, RetiredAt: at, SizeBytes: sizeOf(cell(row, 1)),
			Why: "renamed aside — by postgres_retire_database, or by hand", Kind: KindSetAside})
	}
	return inv, nil
}

// DeleteRetired drops what a retired entry names: a consumer's database and then its role, or one
// database set aside. Answers the bytes freed.
func (c *Client) DeleteRetired(ctx context.Context, r harness.Retired) (int64, error) {
	if r.Kind == KindSetAside {
		if !setAside.MatchString(r.Consumer) {
			return 0, fmt.Errorf("%s is not a database set aside", r.Consumer)
		}
		return c.dropDatabase(ctx, r.Consumer)
	}
	// Only a retired one: the login must be locked here and now, whatever the caller believed.
	row, err := c.Query(ctx, "", "SELECT rolcanlogin FROM pg_roles WHERE rolname = "+Literal(r.Consumer))
	if err != nil {
		return 0, err
	}
	if len(row.Rows) > 0 && cell(row.Rows[0], 0) == "t" {
		return 0, fmt.Errorf("%s can log in — it is active, not retired, and is not deleted", r.Consumer)
	}
	freed, err := c.dropDatabase(ctx, r.Consumer)
	if err != nil {
		return freed, err
	}
	if len(row.Rows) == 0 {
		// Retired by removing the login: its keeper goes with the database it kept.
		keeper, err := KeeperOf(r.Consumer)
		if err != nil {
			return freed, err
		}
		if has, err := c.exists(ctx, "SELECT 1 FROM pg_roles WHERE rolname = "+Literal(keeper)); err != nil || !has {
			return freed, err
		}
		if _, err := c.Query(ctx, "", fmt.Sprintf("DROP OWNED BY %s", Ident(keeper))); err != nil {
			return freed, err
		}
		_, err = c.Query(ctx, "", fmt.Sprintf("DROP ROLE %s", Ident(keeper)))
		return freed, err
	}
	owns, err := c.Query(ctx, "", "SELECT datname FROM pg_database WHERE datdba = (SELECT oid FROM pg_roles WHERE rolname = "+
		Literal(r.Consumer)+")")
	if err != nil {
		return freed, err
	}
	if len(owns.Rows) > 0 {
		return freed, fmt.Errorf("%s's database is dropped; the role is kept because it still owns %s — delete that first",
			r.Consumer, cell(owns.Rows[0], 0))
	}
	_, err = c.Query(ctx, "", fmt.Sprintf("DROP ROLE %s", Ident(r.Consumer)))
	return freed, err
}

func (c *Client) dropDatabase(ctx context.Context, database string) (int64, error) {
	r, err := c.Query(ctx, "", "SELECT pg_database_size(datname) FROM pg_database WHERE datname = "+Literal(database))
	if err != nil || len(r.Rows) == 0 {
		return 0, err
	}
	freed := sizeOf(cell(r.Rows[0], 0))
	if _, err := c.Query(ctx, "", "SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = "+
		Literal(database)+" AND pid <> pg_backend_pid()"); err != nil {
		return 0, err
	}
	if _, err := c.Query(ctx, "", fmt.Sprintf("DROP DATABASE %s", Ident(database))); err != nil {
		return 0, err
	}
	return freed, nil
}

func cell(row []*string, i int) string {
	if i < len(row) && row[i] != nil {
		return *row[i]
	}
	return ""
}

func sizeOf(s string) int64 {
	n, err := strconv.ParseInt(s, 10, 64)
	if err != nil {
		return -1
	}
	return n
}