// A provisioner for Redis, in the form the mesh expects one.
//
// The mesh generated a password, sealed it to the machine that must accept it, and discarded the
// plaintext — so it cannot tell Redis to start accepting it. Something on that machine reads what
// the host wrote and makes it true. This is that something, for the third provision after a
// database and a bucket: a cache.
//
// **It is an example, not part of the control plane** — the same standing as the postgres one,
// whose contract this mirrors line for line:
//
//	$GRANTS/mesh.json          every consumer, what it asked for, and where its credential is
//	$GRANTS/<node>.secret      one consumer's password, alone in the file
//
// What a consumer is given is an ACL user scoped to a key prefix. Redis has no databases to hand
// out — SELECT-numbered ones are deprecated in clusters and shared in spirit — so the unit of
// tenancy is the keyspace pattern: a consumer contributing `prefix: photos` gets a user that can
// touch `photos:*` and nothing else. Administration and the dangerous category stay withheld;
// nothing a tenant is granted can flush the store or read another tenant's keys.
//
// Redis is spoken to in RESP directly, over one connection, with no client library. Five commands
// are needed — AUTH, PING, ACL SETUSER, ACL USERS, ACL DELUSER, ACL SAVE — and a dependency large
// enough to hide what they do would be most of this program's size.
package main

import (
	"bufio"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"net"
	"os"
	"os/signal"
	"path/filepath"
	"sort"
	"strings"
	"syscall"
	"time"
)

// mark is what this provisioner names the users it owns, so it never removes one a person made by
// hand — the mesh's rule about origins, one level down (novox/hq 04-ISSUES/010).
const mark = "mesh_"

type contribution struct {
	From string `json:"from"`
	Node string `json:"node"`
	// As is what to call the user this consumer will authenticate as. Given by the mesh, never
	// invented here (novox/hq 04-ISSUES/023): the consumer has to present it, so both ends must
	// hold the same derivation.
	As     string         `json:"as"`
	Secret string         `json:"secret"`
	Values map[string]any `json:"values"`
}

type manifest struct {
	Requirement string         `json:"requirement"`
	Given       []contribution `json:"given"`
}

func main() {
	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()

	if len(os.Args) > 1 && os.Args[1] == "--watch" {
		if err := watch(ctx); err != nil {
			fmt.Fprintf(os.Stderr, "mesh-provision-redis: %v\n", err)
			os.Exit(1)
		}
		return
	}
	if err := run(ctx); err != nil {
		fmt.Fprintf(os.Stderr, "mesh-provision-redis: %v\n", err)
		os.Exit(1)
	}
}

// watch reconciles now, and again whenever what the mesh delivered changes — by content, not by
// modification time, for the reasons the postgres provisioner records.
func watch(ctx context.Context) error {
	const every = 10 * time.Second
	var last string
	for {
		state, err := given()
		switch {
		case err != nil:
			fmt.Fprintf(os.Stderr, "cannot read what was granted: %v\n", err)
		case state != last:
			if err := run(ctx); err != nil {
				fmt.Fprintf(os.Stderr, "%v\n", err)
			} else {
				last = state
			}
		}
		select {
		case <-ctx.Done():
			return nil
		case <-time.After(every):
		}
	}
}

// given digests everything delivered, so a change of any of it is one comparison and no secret is
// held beyond its hash.
func given() (string, error) {
	entries, err := os.ReadDir(grantsDir())
	if err != nil {
		return "", err
	}
	var names []string
	for _, e := range entries {
		names = append(names, e.Name())
	}
	sort.Strings(names)
	sum := sha256.New()
	for _, name := range names {
		body, err := os.ReadFile(filepath.Join(grantsDir(), name))
		if err != nil {
			return "", err
		}
		fmt.Fprintf(sum, "%s:%x\n", name, sha256.Sum256(body))
	}
	return hex.EncodeToString(sum.Sum(nil)), nil
}

func grantsDir() string {
	if grants := os.Getenv("GRANTS"); grants != "" {
		return grants
	}
	return "/var/lib/redis-module/grants"
}

func run(ctx context.Context) error {
	raw, err := os.ReadFile(filepath.Join(grantsDir(), "mesh.json"))
	if err != nil {
		if os.IsNotExist(err) {
			fmt.Printf("nothing has been granted to this machine\n")
			return nil
		}
		return err
	}
	var m manifest
	if err := json.Unmarshal(raw, &m); err != nil {
		return fmt.Errorf("the manifest at %s is not readable: %w", grantsDir(), err)
	}

	r, err := connect(ctx)
	if err != nil {
		return err
	}
	defer r.Close()

	// Reconciling, not applying a change: the same state from wherever it starts.
	wanted := map[string]bool{}
	for _, c := range m.Given {
		if c.Node == "" {
			continue
		}
		prefix, _ := c.Values["prefix"].(string)
		if prefix == "" {
			return fmt.Errorf("%s asked for a cache and did not say its key prefix", c.From)
		}
		if err := usablePrefix(prefix); err != nil {
			return fmt.Errorf("%s: %w", c.From, err)
		}
		password, err := os.ReadFile(c.Secret)
		if err != nil {
			return fmt.Errorf("%s's credential should be at %s and is not there", c.Node, c.Secret)
		}
		user := c.As
		if user == "" {
			return fmt.Errorf("%s on %s was granted a cache and the mesh did not say what to "+
				"call its user, so there is no name both ends would agree on", c.From, c.Node)
		}
		if !strings.HasPrefix(user, mark) {
			return fmt.Errorf("%s on %s is to be called %q, which does not begin with %q — "+
				"withdrawal finds this provisioner's work by that prefix, so it would never let "+
				"this one go", c.From, c.Node, user, mark)
		}
		wanted[user] = true

		// One SETUSER, from reset: the whole grant every time, so a rotation replaces the
		// password and a changed prefix replaces the keyspace, with no residue of what was.
		if _, err := r.do("ACL", "SETUSER", user, "reset", "on",
			">"+strings.TrimSpace(string(password)),
			"~"+prefix+":*", "&"+prefix+":*",
			"+@all", "-@admin", "-@dangerous"); err != nil {
			return fmt.Errorf("granting %s: %w", user, err)
		}
		fmt.Printf("granted %s the keyspace %s:*\n", user, prefix)
	}

	// And every user this provisioner made that nobody asks for any more — the half usually
	// missing, without which a departed consumer keeps a working login for ever.
	users, err := r.strings("ACL", "USERS")
	if err != nil {
		return err
	}
	for _, user := range users {
		if !strings.HasPrefix(user, mark) || wanted[user] {
			continue
		}
		if _, err := r.do("ACL", "DELUSER", user); err != nil {
			return fmt.Errorf("revoking %s: %w", user, err)
		}
		fmt.Printf("revoked %s\n", user)
	}

	// Persisted, or the next restart forgets every grant. The server runs with an aclfile for
	// exactly this; a server without one refuses the save, and saying so beats a cache that
	// silently loses its tenants on restart.
	if _, err := r.do("ACL", "SAVE"); err != nil {
		return fmt.Errorf("the grants were applied and could not be persisted — a restart would "+
			"forget them: %w", err)
	}
	return nil
}

// resp is the five commands this needs, spoken directly.
type resp struct {
	conn net.Conn
	in   *bufio.Reader
}

func connect(ctx context.Context) (*resp, error) {
	where := os.Getenv("MESH_PROVISION_REDIS")
	if where == "" {
		where = "127.0.0.1:6379"
	}
	var d net.Dialer
	conn, err := d.DialContext(ctx, "tcp", where)
	if err != nil {
		return nil, err
	}
	r := &resp{conn: conn, in: bufio.NewReader(conn)}

	file := os.Getenv("MESH_PROVISION_PASSWORD_FILE")
	if file == "" {
		return nil, fmt.Errorf("MESH_PROVISION_PASSWORD_FILE is not set, and an unauthenticated " +
			"provisioner would mean an unauthenticated store")
	}
	password, err := os.ReadFile(file)
	if err != nil {
		return nil, err
	}
	if _, err := r.do("AUTH", strings.TrimSpace(string(password))); err != nil {
		return nil, fmt.Errorf("the store did not accept the password the mesh sealed here: %w", err)
	}
	if _, err := r.do("PING"); err != nil {
		return nil, err
	}
	return r, nil
}

func (r *resp) Close() { _ = r.conn.Close() }

// do sends one command and returns the reply, flattened to a string for the simple kinds.
func (r *resp) do(args ...string) (any, error) {
	var out strings.Builder
	fmt.Fprintf(&out, "*%d\r\n", len(args))
	for _, a := range args {
		fmt.Fprintf(&out, "$%d\r\n%s\r\n", len(a), a)
	}
	if _, err := r.conn.Write([]byte(out.String())); err != nil {
		return nil, err
	}
	return r.read()
}

// strings is do, for the replies that are arrays of bulk strings.
func (r *resp) strings(args ...string) ([]string, error) {
	reply, err := r.do(args...)
	if err != nil {
		return nil, err
	}
	items, ok := reply.([]any)
	if !ok {
		return nil, fmt.Errorf("expected an array and the store said %v", reply)
	}
	var out []string
	for _, item := range items {
		if s, ok := item.(string); ok {
			out = append(out, s)
		}
	}
	return out, nil
}

func (r *resp) read() (any, error) {
	line, err := r.in.ReadString('\n')
	if err != nil {
		return nil, err
	}
	line = strings.TrimRight(line, "\r\n")
	if line == "" {
		return nil, fmt.Errorf("the store sent an empty reply")
	}
	body := line[1:]
	switch line[0] {
	case '+', ':':
		return body, nil
	case '-':
		// The store's own words, verbatim: it says exactly what it refused and why.
		return nil, fmt.Errorf("%s", body)
	case '$':
		if body == "-1" {
			return nil, nil
		}
		var n int
		fmt.Sscanf(body, "%d", &n)
		buf := make([]byte, n+2)
		if _, err := readFull(r.in, buf); err != nil {
			return nil, err
		}
		return string(buf[:n]), nil
	case '*':
		var n int
		fmt.Sscanf(body, "%d", &n)
		items := make([]any, 0, n)
		for range n {
			item, err := r.read()
			if err != nil {
				return nil, err
			}
			items = append(items, item)
		}
		return items, nil
	}
	return nil, fmt.Errorf("the store began a reply with %q, which this does not speak", line[0])
}

func readFull(in *bufio.Reader, buf []byte) (int, error) {
	total := 0
	for total < len(buf) {
		n, err := in.Read(buf[total:])
		if err != nil {
			return total, err
		}
		total += n
	}
	return total, nil
}

// usablePrefix refuses a prefix that would grant more keyspace than anyone read in the manifest.
//
// The grant is written into an ACL pattern, so a prefix carrying pattern characters stops meaning
// itself: `pho*` granted as `pho*:*` matches every tenant whose name starts with pho. The grant
// must mean what it says, so anything Redis would read as a pattern is refused rather than
// escaped — escaping would create a second naming scheme the mesh does not know about.
func usablePrefix(prefix string) error {
	if prefix == "" {
		return fmt.Errorf("the key prefix is empty, which would grant the whole keyspace")
	}
	if strings.ContainsAny(prefix, " \t\n\r*?[]^{}") {
		return fmt.Errorf("the key prefix %q contains characters Redis reads as a pattern, so "+
			"the grant would match more than it names", prefix)
	}
	return nil
}