package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"os"
	"os/signal"
	"strconv"
	"strings"
	"syscall"
	"time"

	"sync"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"

	"git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
	"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient"
	"git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
	occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
	oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
	catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/sourcebranches"
	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"git.novox.be/novox/mesh-sdk/go/process"
	"git.novox.be/novox/mesh-sdk/go/seatdef"
	"git.novox.be/novox/mesh-sdk/go/seatlist"
	"git.novox.be/novox/mesh-sdk/go/views"

	"git.novox.be/novox/mesh-sdk/go/bussnap"
	"nats-tools/internal/keeper"
)

// The keeper (`nats-tools keep`): the bus module's process on its own account (own secret broker-keeper, ADR 0436 §1),
// started by the node-engine as the manifest's process keeper. It reads the fill every minute from its start, follows the
// catalog half of every watched branch through meshcatalogclient's branch follower and the seat list (design 47 G §344,
// §433, §532, §614), plans again at each of their changes, hears the operator's decided on its durable consumer, and acts
// on the persons' requests its tools put in its own state requests (requests.go) as they are written or answered. It answers the seat mesh-broker's snapshot verbs itself (snapshots.go; design 47 G A-107).

// The router's records of asks, read under this module's own name (ADR 0259 §3), as the mail sender reads them.
const asksBucket = "messenger_asks"

// WarrantFrom reads the operator's warrant on one of the bus module's asks from the router's own record of it, under the
// keeper's own account (read: process.Conn.Ask): the ask and its warrant, or nil while the operator has not answered.
func WarrantFrom(read func(bucket, id string) (*stdio.StateEntry, error)) func(id string) (*oc.Ask, *oc.Warrant, error) {
	return func(id string) (*oc.Ask, *oc.Warrant, error) {
		e, err := read(asksBucket, id)
		if err != nil || e == nil {
			return nil, nil, err
		}
		var rec struct {
			Ask     *oc.Ask     `json:"ask"`
			Warrant *oc.Warrant `json:"warrant"`
		}
		if err := json.Unmarshal(e.Value, &rec); err != nil {
			return nil, nil, err
		}
		return rec.Ask, rec.Warrant, nil
	}
}

// processStore is the keeper's own state through its process connection.
type processStore struct{ c *process.Conn }

func (s processStore) Get(state, key string) ([]byte, uint64, bool, error) {
	e, err := s.c.State(state).Get(key)
	if err != nil || e == nil {
		return nil, 0, false, err
	}
	return e.Value, e.Revision, true, nil
}

func (s processStore) Put(state, key string, value []byte) (uint64, error) {
	return s.c.State(state).Put(key, json.RawMessage(value))
}

func (s processStore) Delete(state, key string) error { return s.c.State(state).Delete(key) }

func (s processStore) Keys(state string) ([]string, error) { return s.c.State(state).Keys() }

func (s processStore) Create(state, key string, value []byte) (uint64, error) {
	rev, err := s.c.State(state).Create(key, json.RawMessage(value))
	if errors.Is(err, stdio.ErrChanged) {
		return 0, keeper.ErrExists
	}
	return rev, err
}

// streamEmitter says an event of the seat mesh-broker, with the SDK's headers, under its one event id (the lead's ruling,
// registry milestone, 13:0x; design 25 § 12.3): on the seat's own subject `mesh.seat.mesh-broker.event.<event>`, which the
// shared events stream keeps 7 days and every consumer of `mesh-broker.<event>` hears, like any seat's event; then on the
// bus module's own stream (a stream that keeps its history, 256 MiB), which its tool `history` reads. The seat's subject
// first: a reader hears it before the history holds it, never the other way round. It says it on s.Left (every subject
// when nil) and leaves in s.Left what the bus did not take, so the keeper says that again with the same id and the bus
// keeps it once (the two publishes are not one act: a failed second one is kept and said again, never silent).
func streamEmitter(js jetstream.JetStream, a process.Account) func(*keeper.Saying) error {
	return func(s *keeper.Saying) error {
		var body any = s.Body
		if s.Restored != nil {
			body = s.Restored
		}
		data, err := json.Marshal(body)
		if err != nil {
			return err
		}
		if s.ID == "" {
			s.ID = keeper.EventID()
		}
		at := s.At
		if at.IsZero() {
			at = time.Now().UTC()
		}
		subjects := s.Left
		if subjects == nil {
			subjects = EventSubjects(a, s.Event)
			if s.Restored != nil {
				// snapshot-restored has its own subject, under the request and the act (§681).
				subjects = []string{meshbroker.SubjectSnapshotRestored(s.Restored.Request, s.Restored.Act)}
			}
		}
		s.Said = nil
		for i, subject := range subjects {
			msg := nats.NewMsg(subject)
			msg.Header.Set("x-event-id", s.ID)
			msg.Header.Set("x-source", a.Module)
			msg.Header.Set("x-node", a.Node)
			msg.Header.Set("x-time", at.UTC().Format(time.RFC3339Nano))
			msg.Header.Set("content-type", "application/json")
			msg.Data = data
			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
			_, err = js.PublishMsg(ctx, msg, jetstream.WithMsgID(s.ID))
			cancel()
			if err != nil {
				s.Left = append([]string(nil), subjects[i:]...)
				return fmt.Errorf("saying %s on %s: %w", s.Event, subject, err)
			}
			s.Said = append(s.Said, subject)
		}
		s.Left = nil
		return nil
	}
}

// EventSubjects are where one event of the seat mesh-broker is said, in order: the seat's subject, then the bus module's
// own stream.
func EventSubjects(a process.Account, event string) []string {
	return []string{seatdef.EventSubject(meshbroker.Seat, event), a.StreamSubject(event)}
}

// sayer raises and clears the keeper's conditions through the condition keeper, as every module says them.
func sayer(js jetstream.JetStream) func(oc.Condition) error {
	return func(c oc.Condition) error {
		subject, err := occ.ConditionSubject(keeper.Module)
		if err != nil {
			return err
		}
		data, err := occ.EncodeCondition(c)
		if err != nil {
			return err
		}
		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		_, err = js.Publish(ctx, subject, data)
		return err
	}
}

func percent(getenv func(string) string, name string, fallback int64) int64 {
	if n, err := strconv.ParseInt(strings.TrimSpace(getenv(name)), 10, 64); err == nil && n > 0 && n < 100 {
		return n
	}
	return fallback
}

// runKeeper is the keeper's process: it returns when ctx ends, or when it has no credential.
func runKeeper(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
	logger := log.New(logOut, "nats-keeper: ", log.LstdFlags)
	settings := readSettings(getenv("MESH_NATS_SNAPSHOTS_DIR") + "/settings.json")
	account, err := process.ReadBrokerAccount(getenv)
	if err != nil {
		return err
	}
	conn, err := process.Connect(account)
	if err != nil {
		return err
	}
	defer conn.Close()
	nc, js, err := account.Dial(nats.Name(account.Node + "." + account.Module + ".keeper"))
	if err != nil {
		return err
	}
	defer nc.Close()
	recorder := meshhandactsclient.Recorder{Ask: conn.Call}
	// The bus module's findings go through a held writer over its own state findings (design 47 G §681, A-29, A-40), made at once
	// and started before the keeper's loop, so what the keeper writes first is held and written at the start.
	writer, err := views.NewHeldWriter(views.WriterConfig[meshconditions.Finding, meshconditions.FindingCleared, meshconditions.FindingsStated, meshconditions.FindingsAskedEvent]{
		Asked: meshconditions.SubjectFindingsAsked, Module: keeper.Module, Node: account.Node, State: conn.State("findings"),
		Kinds: []string{keeper.FindingSnapshotsFilling}, JS: js, Conn: nc, Logf: logger.Printf})
	if err != nil {
		return err
	}
	if err := writer.Start(ctx); err != nil {
		return err
	}
	defer writer.Close()
	k := &keeper.Keeper{
		Bus: keeper.JetStream{JS: js, Snap: bussnap.Into(nc, getenv("MESH_NATS_SNAPSHOTS_DIR")), NC: nc,
			Dir: getenv("MESH_NATS_SNAPSHOTS_DIR"), Log: logger.Printf, Server: serverStorageLimit(LiveMonitor())},
		Store: processStore{conn}, Emit: streamEmitter(js, account), Say: sayer(js),
		Act: func(what, why, cause string) error {
			_, err := recorder.Record(meshhandactsclient.Request{What: what, Why: why, Cause: cause})
			return err
		},
		Warrant: WarrantFrom(conn.Ask), Finding: writer,
		Logf:  logger.Printf,
		Raise: percent(getenv, "MESH_NATS_FULL_RAISE", 75), Clear: percent(getenv, "MESH_NATS_FULL_CLEAR", 70),
		// The settings snapshots-cap (20 GiB) and snapshot-keep-days (90) of the manifest, read from the file it declares
		// at the start, its defaults when the file is missing (design 47 G §681, A-112).
		SnapshotsCap:     settings.SnapshotsCap,
		SnapshotKeepDays: settings.SnapshotKeepDays,
	}
	// The catalog half (design 47 G §433, §532, §614): meshcatalogclient's branch follower on the keeper's views reader, which
	// holds its finding writer; and the seat list, each definition as the seat answers that one seat, resolved (§537). Each
	// change of a branch's view, of the registrations or of the seat list plans again.
	moved := make(chan struct{}, 1)
	wakeMoved := func() {
		select {
		case moved <- struct{}{}:
		default:
		}
	}
	follow, err := followCatalog(ctx, conn, writer, wakeMoved, logger.Printf)
	if err != nil {
		return err
	}
	k.Catalog, k.Seats, k.Again = follow.view, follow.seats, follow.again
	// The process answers the seat mesh-broker's snapshot verbs itself (design 47 G §681, A-107), with the caller the bus proved.
	stopAnswers, err := conn.AnswerSeat(meshbroker.Seat, seatlist.Definition(func() *seatlist.Reader { return follow.seatList }, meshbroker.Seat), snapshotAnswers(k, heldRequest(k.Store), keeperSnapshots(nc, logger.Printf)))
	if err != nil {
		return fmt.Errorf("the seat %s's snapshot verbs cannot be answered: %w", meshbroker.Seat, err)
	}
	defer stopAnswers()
	logger.Printf("started on the account %s: following the catalog half of every watched branch through the seat %s", account.User, catalog.Seat)
	return Keep(ctx, k, hearer(js, account, follow.take, follow.bound, logger.Printf), requestsOf(conn), moved, logger.Printf)
}

// catalogFollow is the keeper's reading of the catalog half and the seat list (design 47 G §532, §614): the reader, the
// branch follower over it, the seat list, and the events of both, routed from the keeper's one durable consumer.
type catalogFollow struct {
	ctx      context.Context
	reader   *views.ProcessReader
	branches *meshcatalogclient.Branches
	seatList *seatlist.Reader
	routes   *eventRoutes
}

// followCatalog wires the branch follower and the seat list, and asks their first views; moved is told of each change.
func followCatalog(ctx context.Context, conn *process.Conn, writer views.WireWriter, moved func(), logf func(string, ...any)) (*catalogFollow, error) {
	f := &catalogFollow{ctx: ctx, routes: &eventRoutes{by: map[string]func(jetstream.Msg){}}}
	f.reader = &views.ProcessReader{Module: keeper.Module, Call: conn.Call, Logf: logf, Findings: func() views.WireWriter { return writer }}
	branches, err := meshcatalogclient.FollowBranchesProcess(f.reader, conn.Call)
	if err != nil {
		return nil, fmt.Errorf("the catalog half cannot be followed: %w", err)
	}
	branches.Logf = logf
	branches.Changed = func(string, string, json.RawMessage, bool, time.Time) { moved() }
	// A registration that changes its watched branches' order moves what is planned from which branch (§532).
	reconcile := branches.Registrations.Follower.Changed
	branches.Registrations.Follower.Changed = func(key string, value json.RawMessage, deleted bool, at time.Time) {
		reconcile(key, value, deleted, at)
		moved()
	}
	f.branches = branches
	f.seatList, err = seatlist.Start(seatlist.Config{Bus: seatlist.ProcessBus(conn.Call, f.routes), Bootstrap: seatlist.Bootstrapped(),
		OnChange: func([]catalogue.Seat) { moved() }})
	if err != nil {
		logf("the seat list cannot be followed (%v): a module's sent store waits for it", err)
	}
	f.reader.Ask(ctx, nil)
	return f, nil
}

// view is the catalog half as the keeper plans from it: an error while a watched branch has no first view yet.
func (f *catalogFollow) view() (keeper.CatalogView, error) {
	if err := f.branches.Ready(); err != nil {
		return keeper.CatalogView{}, err
	}
	var bs []Branch
	for _, name := range f.branches.Followed() {
		if c, ok := f.branches.Catalog(name); ok {
			if half, ok := c.View(); ok {
				bs = append(bs, Branch{Name: name, Revision: half.Revision.Revision, Entries: half.Entries})
			}
		}
	}
	return ViewOf(bs, func(module string) []string {
		reg, ok := f.branches.Registrations.Registration(module)
		if !ok {
			return nil
		}
		return reg.Branches
	})
}

// seats are the seat list's definitions once the seat has answered; before, none, so no sent store is planned yet
// (ADR 0636 §1: a reader that has not heard acts on nothing new).
func (f *catalogFollow) seats() map[string]seatdef.Definition {
	if f.seatList == nil || !f.seatList.Status().Heard {
		return nil
	}
	return f.seatList.Definitions()
}

// again asks every first view of the catalog half and the seat list again, by hand.
func (f *catalogFollow) again() error {
	var errs []error
	for _, fl := range f.reader.Followers() {
		errs = append(errs, fl.Again(f.ctx, "asked again by hand (nats_first_view_again)"))
	}
	if f.seatList != nil {
		errs = append(errs, f.seatList.Reload())
	}
	return errors.Join(errs...)
}

// take hands a message of the catalog half or the seat list to its reader, which acknowledges it; false for any other.
func (f *catalogFollow) take(m jetstream.Msg) bool {
	if h := f.routes.get(m.Subject()); h != nil {
		h(m)
		return true
	}
	if strings.HasPrefix(m.Subject(), "mesh.seat."+catalog.Seat+".event.") || strings.HasPrefix(m.Subject(), seatdef.EventSubject(sourcebranches.Seat, sourcebranches.EventSourceMoved)) {
		f.reader.Hear(f.ctx)(m)
		return true
	}
	return false
}

// bound tells each view the bus made the keeper's consumer then, so a first view taken before is asked again.
func (f *catalogFollow) bound(created time.Time) {
	for _, fl := range f.reader.Followers() {
		fl.ConsumerMade(f.ctx, created)
	}
}

// eventRoutes are the seat list's handlers of its events by subject, on the keeper's one durable consumer.
type eventRoutes struct {
	mu sync.Mutex
	by map[string]func(jetstream.Msg)
}

func (r *eventRoutes) Handle(subject string, handle func(jetstream.Msg)) error {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.by[subject] != nil {
		return fmt.Errorf("the events %q are handled already", subject)
	}
	r.by[subject] = handle
	return nil
}

func (r *eventRoutes) get(subject string) func(jetstream.Msg) {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.by[subject]
}

// hearer binds the keeper's durable consumer on EVENTS (<node>_nats_keeper, filtered on its consumes): a message of the
// catalog half or the seat list goes to take, which acknowledges it; every other to handle, acknowledged once applied.
// bound is told when the bus made the consumer.
func hearer(js jetstream.JetStream, a process.Account, take func(jetstream.Msg) bool, bound func(time.Time), logf func(string, ...any)) func(context.Context, func(subject string, body []byte) error) error {
	return func(ctx context.Context, handle func(string, []byte) error) error {
		cons, err := js.Consumer(ctx, process.EventsStream, a.QueueConsumerName())
		if err != nil {
			return fmt.Errorf("the keeper's consumer %s on %s: %w", a.QueueConsumerName(), process.EventsStream, err)
		}
		if info, err := cons.Info(ctx); err == nil && bound != nil {
			bound(info.Created.UTC())
		}
		cc, err := process.HearEvents(cons, func(m jetstream.Msg) {
			if take != nil && take(m) {
				return
			}
			if err := handle(m.Subject(), m.Data()); err != nil {
				logf("%v: delivered again", err)
				_ = m.NakWithDelay(30 * time.Second)
				return
			}
			_ = m.Ack()
		}, logf)
		if err != nil {
			return err
		}
		go func() { <-ctx.Done(); cc.Stop() }()
		return nil
	}
}

// FillEvery is how often the keeper's fill pass runs: a minute (a test's shorter bound beside).
var FillEvery = time.Minute

// DecidedSubject is where the operator channel says the operator's answer to one of the bus module's own asks (ADR
// 0259 §6; ADR 0620 §2), heard on the keeper's durable consumer: the warrant a recreate or a drop waits for.
var DecidedSubject = oc.SubjectDecided(keeper.Module)

// Keep is the keeper's loop, apart from the bus so a test drives it: a fill pass at once and every minute, from its
// start, so the store's one guard runs while the catalog half's first views are still asked (design 25 § 12.3); a plan
// from the catalog half at once and at each move of it (moved: a branch's view, the registrations or the seat list
// changed), nothing planned until every watched branch has its first view (design 47 G §532, §614); hear, which wakes the
// persons' requests on the operator channel's decided for the bus module's own asks; the waiting entries made again on
// a backoff; and the persons' requests acted on as they are written or answered, never on a timer. It returns when ctx
// ends.
func Keep(ctx context.Context, k *keeper.Keeper, hear func(context.Context, func(string, []byte) error) error,
	requests func(context.Context, func()) error, moved <-chan struct{}, logf func(string, ...any)) error {
	tick := time.NewTicker(FillEvery)
	defer tick.Stop()
	k.FillPass()
	k.SnapshotsAtStart()
	k.EndCutOffRestores()
	plan := func() {
		if err := k.PlanCatalog(); err != nil {
			logf("nothing is planned yet: %v", err)
		}
	}
	plan()
	wake := func(c chan struct{}) func() {
		return func() {
			select {
			case c <- struct{}{}:
			default:
			}
		}
	}
	asked := make(chan struct{}, 1)
	for backoff := 10 * time.Second; ; backoff = min(backoff*2, 5*time.Minute) {
		err := hear(ctx, func(subject string, body []byte) error {
			if subject == DecidedSubject {
				// The operator answered one of the bus module's asks: the waiting requests are read again now.
				wake(asked)()
			}
			return nil
		})
		if err == nil {
			break
		}
		logf("the keeper's consumer cannot be heard yet (%v): asked again in %s", err, backoff)
		// The fill pass and the plans run on while the consumer is bound again (review round 2, running F1).
		bound := time.After(backoff)
	binding:
		for {
			select {
			case <-ctx.Done():
				return ctx.Err()
			case <-tick.C:
				k.FillPass()
			case <-moved:
				plan()
			case <-bound:
				break binding
			}
		}
	}
	if requests != nil {
		if err := requests(ctx, wake(asked)); err != nil {
			logf("the persons' requests cannot be watched (%v): acted on as the operator answers", err)
		}
	}
	// What waited while the keeper was down, or before its first view, is acted on now.
	k.ActOnRequests()
	retry := 10 * time.Second
	var again <-chan time.Time
	for {
		if len(k.Waiting()) > 0 && again == nil {
			again = time.After(retry)
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-tick.C:
			k.FillPass()
		case <-asked:
			k.ActOnRequests()
		case <-moved:
			plan()
		case <-again:
			again = nil
			k.CheckAll()
			if len(k.Waiting()) > 0 {
				retry = min(retry*2, 5*time.Minute)
				logf("%d entr(y/ies) wait to be made: %s; made again in %s", len(k.Waiting()), strings.Join(k.Waiting(), "; "), retry)
			} else {
				retry = 10 * time.Second
			}
		}
	}
}

// keepMain runs the keeper until a signal.
func keepMain() {
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()
	if err := runKeeper(ctx, os.Getenv, os.Stderr); err != nil && ctx.Err() == nil {
		fmt.Fprintln(os.Stderr, "nats-keeper:", err)
		os.Exit(1)
	}
}

// requestsOf watches the keeper's own state requests, waking act at each change.
func requestsOf(c *process.Conn) func(context.Context, func()) error {
	return func(_ context.Context, act func()) error {
		return c.State(keeper.StateRequests).Watch("", func(stdio.StateChange) error {
			act()
			return nil
		})
	}
}

// keeperSettings are the keeper's settings its manifest declares as a file (design 47 G A-112): the content of
// `${config:<key>}`, read at its start.
type keeperSettings struct {
	SnapshotsCap     int64 `json:"snapshots-cap"`
	SnapshotKeepDays int   `json:"snapshot-keep-days"`
}

// readSettings reads the settings file at path; a file that is missing, or a value that is not above 0, gives its default.
func readSettings(path string) keeperSettings {
	s := keeperSettings{SnapshotsCap: keeper.DefaultSnapshotsCap, SnapshotKeepDays: keeper.DefaultSnapshotKeepDays}
	raw, err := os.ReadFile(path)
	if err != nil {
		return s
	}
	var f keeperSettings
	if json.Unmarshal(raw, &f) != nil {
		return s
	}
	if f.SnapshotsCap > 0 {
		s.SnapshotsCap = f.SnapshotsCap
	}
	if f.SnapshotKeepDays > 0 {
		s.SnapshotKeepDays = f.SnapshotKeepDays
	}
	return s
}