package hooks

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"net"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"sourcecraft.dev/bigbes/sr-ht-spec/authn"
	"sourcecraft.dev/bigbes/sr-ht-spec/core"
	"sourcecraft.dev/bigbes/sr-ht-spec/gitx"
	"sourcecraft.dev/bigbes/sr-ht-spec/service"
)

// Backend is the daemon this package fronts. *service.Service satisfies it as
// written; it is an interface so the receive path can be exercised end to end,
// against a real `git push`, without a Postgres instance.
//
// Everything policy-shaped lives behind ValidatePush. This package decides
// which space a repository is, who is pushing, and whether validation was
// waived — and then asks. That split is the point of the whole design: the API
// and the push path must not be able to disagree about what is valid.
type Backend interface {
	// ReposRoot is [spec.sr.ht] repos, the only directory a hook may address a
	// repository under.
	ReposRoot() string

	// Resolver carries the instance owner username, the one identity an
	// "owner" credential can resolve to.
	Resolver() *authn.Resolver

	// TokenStore is what an agent's token is validated against.
	TokenStore() *service.TokenStore

	// ValidatePush answers whether one ref may move. A *service.PushRejection
	// is a policy refusal whose Error() is the text to print; anything else is
	// an infrastructure failure and must fail the push closed.
	ValidatePush(ctx context.Context, req service.PushRequest) error
}

// PushNotifier is what the daemon does when a push lands: reindex the changed
// documents and advance the space's index rev stamp.
//
// It is injected rather than being a Backend method because Phase 1 has no
// indexer — bleve and the index rev stamp belong to Phase 2 — and a stamp
// advanced without a reindex behind it would be a lie the reconciler could not
// detect. Until then the daemon supplies a notifier that records the push and
// nothing else, and the reconciler reports the staleness.
type PushNotifier func(ctx context.Context, space core.SpaceRef, updates []RefUpdate) error

// Options configures a Server.
type Options struct {
	// Backend and Socket are required.
	Backend Backend
	Socket  string

	// OnPush is called for each landed push. Required: a server with no
	// notifier would accept post-receive calls and drop them, which reads
	// exactly like a working index that never updates.
	OnPush PushNotifier

	// Log defaults to slog.Default().
	Log *slog.Logger

	// Timeout bounds handling one request. Zero means DefaultTimeout.
	Timeout time.Duration

	// OptionTTL is how long a push's recorded options survive without the
	// matching update calls. Zero means DefaultOptionTTL.
	OptionTTL time.Duration
}

const (
	// DefaultOptionTTL is how long the daemon remembers a push's options. The
	// gap between pre-receive and the last update of one push is milliseconds;
	// this is three orders of magnitude of slack, and the only cost of an
	// entry outliving its push is a few hundred bytes until it is swept.
	DefaultOptionTTL = 10 * time.Minute

	// maxPendingPushes caps the option table. Any local process can connect to
	// the socket, so the table must not be a way to exhaust memory; at the
	// documented volume — one human, tens of documents a day — a thousand
	// pushes in flight at once means something is wrong, and failing closed is
	// the right answer to that.
	maxPendingPushes = 1024

	// socketDirMode keeps the socket's directory private to the service user.
	// The socket is an unauthenticated write path to the daemon: anyone who
	// can connect can assert the owner principal.
	socketDirMode = 0o700
	socketMode    = 0o600
)

// Server is the daemon side of the receive path: a unix socket the hooks call.
//
// The socket is unix-domain and not a TCP port on localhost, for one reason
// that decides it: filesystem permissions. An "owner" credential is an
// assertion, so the ability to connect is the ability to push as the owner —
// a 0700 directory holding a 0600 socket makes that reachable only by the
// service user, which no localhost TCP port can do.
type Server struct {
	backend   Backend
	socket    string
	log       *slog.Logger
	onPush    PushNotifier
	timeout   time.Duration
	optionTTL time.Duration

	listener net.Listener
	wg       sync.WaitGroup

	// closing distinguishes a listener we shut down from one that failed. Both
	// surface as net.ErrClosed out of Accept, and only one of them is an error.
	closing atomic.Bool

	mu      sync.Mutex
	pending map[string]*pendingPush
}

// pendingPush is what pre-receive told us about a push, waiting for the update
// calls it belongs to.
type pendingPush struct {
	updates []RefUpdate
	skip    bool
	expires time.Time
}

// NewServer builds a server. It does not listen; call Listen.
func NewServer(opts Options) (*Server, error) {
	if opts.Backend == nil {
		return nil, errors.New("hooks: no backend")
	}
	if opts.Socket == "" {
		return nil, errors.New("hooks: no socket path")
	}
	if !filepath.IsAbs(opts.Socket) {
		return nil, fmt.Errorf("hooks: socket path %q is not absolute", opts.Socket)
	}
	if opts.OnPush == nil {
		return nil, errors.New("hooks: no push notifier")
	}
	if opts.Backend.ReposRoot() == "" {
		return nil, errors.New("hooks: backend has no repos root")
	}
	log := opts.Log
	if log == nil {
		log = slog.Default()
	}
	timeout := opts.Timeout
	if timeout <= 0 {
		timeout = DefaultTimeout
	}
	ttl := opts.OptionTTL
	if ttl <= 0 {
		ttl = DefaultOptionTTL
	}
	return &Server{
		backend:   opts.Backend,
		socket:    opts.Socket,
		log:       log,
		onPush:    opts.OnPush,
		timeout:   timeout,
		optionTTL: ttl,
		pending:   make(map[string]*pendingPush),
	}, nil
}

// Socket is the path this server listens on.
func (s *Server) Socket() string { return s.socket }

// Listen binds the socket.
//
// A leftover socket file from a crashed daemon is removed, but only after
// proving it is dead: if something answers on it, another daemon is running
// and this one refuses to start rather than stealing the push path from it.
func (s *Server) Listen() error {
	if err := os.MkdirAll(filepath.Dir(s.socket), socketDirMode); err != nil {
		return fmt.Errorf("hooks: create %s: %w", filepath.Dir(s.socket), err)
	}
	// The directory may pre-date this version, or have been created with a
	// looser umask; make it private either way.
	if err := os.Chmod(filepath.Dir(s.socket), socketDirMode); err != nil {
		return fmt.Errorf("hooks: restrict %s: %w", filepath.Dir(s.socket), err)
	}

	if _, err := os.Stat(s.socket); err == nil {
		conn, dialErr := net.DialTimeout("unix", s.socket, time.Second)
		if dialErr == nil {
			conn.Close()
			return fmt.Errorf("hooks: %s is already served by another process", s.socket)
		}
		if err := os.Remove(s.socket); err != nil {
			return fmt.Errorf("hooks: remove the stale socket %s: %w", s.socket, err)
		}
		s.log.Warn("removed a stale hook socket", "socket", s.socket, "dial_error", dialErr)
	} else if !errors.Is(err, os.ErrNotExist) {
		return fmt.Errorf("hooks: stat %s: %w", s.socket, err)
	}

	ln, err := net.Listen("unix", s.socket)
	if err != nil {
		return fmt.Errorf("hooks: listen on %s: %w", s.socket, err)
	}
	if err := os.Chmod(s.socket, socketMode); err != nil {
		ln.Close()
		return fmt.Errorf("hooks: restrict %s: %w", s.socket, err)
	}
	s.listener = ln
	return nil
}

// Serve accepts hook connections until ctx is cancelled, then waits for the
// calls already in flight. Listen must have succeeded first.
func (s *Server) Serve(ctx context.Context) error {
	if s.listener == nil {
		return errors.New("hooks: Serve called before Listen")
	}

	done := make(chan struct{})
	defer close(done)
	go func() {
		select {
		case <-ctx.Done():
			s.listener.Close()
		case <-done:
		}
	}()

	for {
		conn, err := s.listener.Accept()
		if err != nil {
			s.wg.Wait()
			if ctx.Err() != nil || s.closing.Load() {
				return nil
			}
			return fmt.Errorf("hooks: accept on %s: %w", s.socket, err)
		}
		s.wg.Add(1)
		go func() {
			defer s.wg.Done()
			s.serveConn(ctx, conn)
		}()
	}
}

// Close stops listening and removes the socket. It is idempotent, and safe to
// call after Serve has already returned: a daemon shuts down by cancelling
// Serve's context and then calling this, and a daemon that failed to start
// after Listen calls only this.
func (s *Server) Close() error {
	if s.listener == nil {
		return nil
	}
	s.closing.Store(true)
	err := s.listener.Close()
	if errors.Is(err, net.ErrClosed) {
		err = nil
	}
	s.wg.Wait()
	// net's unix listener unlinks the socket itself; removing it again is
	// tolerated so a listener built elsewhere is still cleaned up.
	if rmErr := os.Remove(s.socket); rmErr != nil && !errors.Is(rmErr, os.ErrNotExist) {
		if err == nil {
			err = fmt.Errorf("hooks: remove %s: %w", s.socket, rmErr)
		}
	}
	return err
}

func (s *Server) serveConn(ctx context.Context, conn net.Conn) {
	defer conn.Close()

	ctx, cancel := context.WithTimeout(ctx, s.timeout)
	defer cancel()
	if deadline, ok := ctx.Deadline(); ok {
		// The context bounds the handler, not the socket reads; without a
		// deadline on the connection a peer that connects and says nothing
		// holds a goroutine forever.
		if err := conn.SetDeadline(deadline); err != nil {
			s.log.Error("could not bound a hook connection", "socket", s.socket, "error", err)
			return
		}
	}

	req, err := ReadRequest(conn)
	if err != nil {
		s.log.Error("unreadable hook request", "socket", s.socket, "error", err)
		// The peer may be something that is not a hook at all; answer anyway,
		// so a hook that sent a message we could not parse still gets a
		// refusal rather than a closed connection it has to interpret.
		s.reply(conn, errorResponse("the daemon could not read the request: %v", err))
		return
	}

	resp := s.handle(ctx, req)
	s.reply(conn, resp)
}

func (s *Server) reply(conn net.Conn, resp Response) {
	if err := WriteResponse(conn, resp); err != nil {
		s.log.Error("could not answer a hook", "socket", s.socket, "error", err)
	}
}

// handle answers one request. It never panics a connection into the pusher's
// terminal: every failure becomes a Response the hook knows how to print.
func (s *Server) handle(ctx context.Context, req Request) Response {
	if err := req.Validate(); err != nil {
		s.log.Error("malformed hook request", "method", req.Method, "error", err)
		return errorResponse("malformed %s request: %v", req.Method, err)
	}

	space, err := s.spaceFor(req.Repo)
	if err != nil {
		s.log.Error("hook named a repository we do not own", "repo", req.Repo, "error", err)
		return errorResponse("%v", err)
	}

	principal, resp, ok := s.principal(ctx, space, req.Credential)
	if !ok {
		return resp
	}

	log := s.log.With(
		"space", space.String(),
		"principal", principal.String(),
		"push", req.Push,
		"method", string(req.Method),
	)

	switch req.Method {
	case MethodPushOptions:
		return s.handlePushOptions(req, space, log)
	case MethodValidateRef:
		return s.handleValidateRef(ctx, req, space, principal, log)
	case MethodPushed:
		return s.handlePushed(ctx, req, space, log)
	default:
		// Request.Validate already rejected anything else.
		return errorResponse("unhandled method %q", req.Method)
	}
}

// handlePushOptions records what pre-receive saw, and refuses a push option
// this service does not understand.
func (s *Server) handlePushOptions(req Request, space core.SpaceRef, log *slog.Logger) Response {
	if unknown := UnknownOptions(req.Options); len(unknown) > 0 {
		log.Info("refused unknown push options", "options", unknown)
		return rejectedResponse(unknownOptionMessage(space, unknown))
	}
	skip := SkipValidation(req.Options)
	if err := s.remember(req, skip); err != nil {
		log.Error("could not record push options", "error", err)
		return errorResponse("%v", err)
	}
	log.Info("push received", "refs", len(req.Updates), "skip_validation", skip)
	return okResponse()
}

// handleValidateRef is the whole of the rejecting path: the refs rule, then
// frontmatter and document-id validation, both inside service.ValidatePush so
// the push path and the API cannot drift apart.
func (s *Server) handleValidateRef(ctx context.Context, req Request, space core.SpaceRef,
	principal authn.Principal, log *slog.Logger) Response {

	update := req.Updates[0]
	skip, err := s.recall(req, update)
	if err != nil {
		log.Error("no recorded pre-receive phase for this push", "ref", update.Ref, "error", err)
		return errorResponse("%v", err)
	}

	err = s.backend.ValidatePush(ctx, service.PushRequest{
		Space:          space,
		Principal:      principal,
		Ref:            update.Ref,
		Old:            update.Old,
		New:            update.New,
		SkipValidation: skip,
	})
	var rejection *service.PushRejection
	switch {
	case err == nil:
		log.Info("ref accepted", "ref", update.Ref, "skip_validation", skip)
		return okResponse()
	case errors.As(err, &rejection):
		log.Info("ref rejected", "ref", update.Ref, "problems", len(rejection.Problems),
			"skippable", rejection.Skippable)
		return rejectedResponse(rejection.Error())
	default:
		// Not a policy answer: Postgres down, an unreadable object, a space
		// with no row. The hook fails the push closed on it.
		log.Error("could not validate a ref", "ref", update.Ref, "error", err)
		return errorResponse("spec.sr.ht could not validate %s: %v", update.Ref, err)
	}
}

// handlePushed notifies the daemon that refs moved. Its answer cannot stop
// anything — git ignores post-receive's exit status — but it is still reported
// honestly so the hook can warn that the index is stale.
func (s *Server) handlePushed(ctx context.Context, req Request, space core.SpaceRef, log *slog.Logger) Response {
	s.forget(req)
	if err := s.onPush(ctx, space, req.Updates); err != nil {
		log.Error("could not record a landed push", "refs", len(req.Updates), "error", err)
		return errorResponse("%v", err)
	}
	log.Info("push landed", "refs", len(req.Updates))
	return okResponse()
}

// spaceFor turns the repository a hook is running in into a space.
//
// The hook's claim is never taken at face value: the path is matched against
// this daemon's own repos root and then re-derived through gitx.DiskPath, so
// the only paths that resolve are the ones this daemon would itself have
// created. gitx is the single source of truth for that layout — deriving it
// twice is how the two copies drift.
func (s *Server) spaceFor(repo string) (core.SpaceRef, error) {
	root, err := filepath.EvalSymlinks(s.backend.ReposRoot())
	if err != nil {
		return core.SpaceRef{}, fmt.Errorf("the repos root %s is unreadable: %w",
			s.backend.ReposRoot(), err)
	}
	root, err = filepath.Abs(root)
	if err != nil {
		return core.SpaceRef{}, fmt.Errorf("the repos root %s is unresolvable: %w",
			s.backend.ReposRoot(), err)
	}

	clean := filepath.Clean(repo)
	rel, err := filepath.Rel(root, clean)
	if err != nil {
		return core.SpaceRef{}, fmt.Errorf("%s is not under the repos root %s", repo, root)
	}
	segs := strings.Split(rel, string(filepath.Separator))
	if len(segs) != 2 || segs[0] == ".." || !strings.HasPrefix(segs[0], "~") {
		return core.SpaceRef{}, fmt.Errorf("%s is not a space repository; "+
			"this daemon serves %s/~<owner>/<space> only", repo, root)
	}
	ref, err := core.ParseSpaceRef(rel)
	if err != nil {
		return core.SpaceRef{}, fmt.Errorf("%s does not name a valid space: %w", repo, err)
	}
	if want := gitx.DiskPath(root, ref); want != clean {
		return core.SpaceRef{}, fmt.Errorf("%s does not name a space; %s would live at %s",
			repo, ref, want)
	}
	return ref, nil
}

// principal resolves the credential a hook forwarded.
//
// The owner is not looked up: sshd authenticated the SSH key and the forced
// command asserted it, and there is exactly one owner on this instance, so the
// name comes from the resolver rather than from the wire — a hook cannot name
// somebody else. An agent's token is checked on every push.
func (s *Server) principal(ctx context.Context, space core.SpaceRef, cred Credential) (authn.Principal, Response, bool) {
	owner := s.backend.Resolver().Owner()
	switch cred.Kind {
	case PrincipalOwner:
		return authn.Principal{Kind: authn.KindOwner, Owner: owner, CookieUser: owner}, Response{}, true
	case PrincipalAgent:
		tok, err := authn.ResolveAgentToken(ctx, s.backend.TokenStore(), cred.Token)
		if err != nil {
			if authn.IsAuthFailure(err) {
				s.log.Warn("refused an agent push", "space", space.String(), "error", err)
				return authn.Principal{}, rejectedResponse(badTokenMessage(space, err)), false
			}
			// The store could not answer. That is not a bad credential and
			// must not read as one; fail the push closed instead.
			s.log.Error("could not validate an agent token", "space", space.String(), "error", err)
			return authn.Principal{}, errorResponse(
				"spec.sr.ht could not check the agent token presented with this push: %v", err), false
		}
		// The local plane, and only it: a push arrives over SSH with a token in
		// a hook's environment, and this path checks it against agent_token and
		// nothing else. A tokens.sr.ht working token is not accepted here — the
		// two planes meet in authn.Resolver, which serves the HTTP surfaces.
		return authn.Principal{
			Kind:      authn.KindAgent,
			Owner:     owner,
			Agent:     cred.Agent,
			Session:   cred.Session,
			TokenName: tok.Name,
			Plane:     authn.PlaneLocal,
		}, Response{}, true
	default:
		// Request.Validate rejected every other spelling already.
		return authn.Principal{}, errorResponse("unknown principal kind %q", cred.Kind), false
	}
}

// pendingKey identifies one push: the repository plus the pid of the
// receive-pack process every hook of that push is a child of.
func pendingKey(req Request) string { return req.Repo + "\x00" + req.Push }

// remember stores what pre-receive saw. It sweeps expired entries first, and
// refuses rather than growing without bound.
func (s *Server) remember(req Request, skip bool) error {
	now := time.Now()
	s.mu.Lock()
	defer s.mu.Unlock()
	s.sweepLocked(now)
	if len(s.pending) >= maxPendingPushes {
		return fmt.Errorf("spec.sr.ht is already tracking %d pushes in flight and cannot accept another",
			len(s.pending))
	}
	s.pending[pendingKey(req)] = &pendingPush{
		updates: append([]RefUpdate(nil), req.Updates...),
		skip:    skip,
		expires: now.Add(s.optionTTL),
	}
	return nil
}

// recall answers whether validation was waived for this ref.
//
// The pre-receive record is required, not optional. Absence is not read as
// "not waived": it means either that the repository's hooks are half installed
// — no pre-receive, so no push option would ever be seen and skip-validation
// would silently never work — or that the daemon restarted mid-push. Both
// deserve a sentence rather than a guess.
//
// The recorded ref update must also match this one exactly. That is what makes
// the pid safe as a correlation key: a recycled pid would have to be paired
// with an identical ref, old and new object name to be mistaken for this push.
func (s *Server) recall(req Request, update RefUpdate) (bool, error) {
	now := time.Now()
	s.mu.Lock()
	defer s.mu.Unlock()
	s.sweepLocked(now)

	entry, ok := s.pending[pendingKey(req)]
	if !ok {
		return false, fmt.Errorf("the daemon did not see the pre-receive phase of this push. "+
			"Either the repository's hooks are only partly installed (all of %s must be present) "+
			"or the daemon restarted mid-push; push again",
			strings.Join(modeNames(), ", "))
	}
	for _, u := range entry.updates {
		if u == update {
			return entry.skip, nil
		}
	}
	return false, fmt.Errorf("the pre-receive phase of push %s did not announce %s; "+
		"the daemon will not validate a ref it was not told about", req.Push, update)
}

// forget drops a push's record once post-receive has run.
func (s *Server) forget(req Request) {
	s.mu.Lock()
	defer s.mu.Unlock()
	delete(s.pending, pendingKey(req))
	s.sweepLocked(time.Now())
}

func (s *Server) sweepLocked(now time.Time) {
	for k, v := range s.pending {
		if now.After(v.expires) {
			delete(s.pending, k)
		}
	}
}

func modeNames() []string {
	out := make([]string, 0, len(Modes()))
	for _, m := range Modes() {
		out = append(out, string(m))
	}
	return out
}

// unknownOptionMessage is what a mistyped push option prints. It is a
// rejection rather than a shrug because there is exactly one option in the
// vocabulary: silently ignoring "--push-option=skip-validaton" would reject
// the push for the very thing the human believed they had waived.
func unknownOptionMessage(space core.SpaceRef, unknown []string) string {
	var b strings.Builder
	fmt.Fprintf(&b, "spec.sr.ht rejected this push.\n\n")
	fmt.Fprintf(&b, "  space: %s\n\n", space)
	for _, o := range unknown {
		fmt.Fprintf(&b, "  --push-option=%s is not a push option this service knows\n", o)
	}
	fmt.Fprintf(&b, "\nThe only push option is --push-option=%s, which waives\n", OptionSkipValidation)
	fmt.Fprintf(&b, "frontmatter and document-id validation. Nothing was written.\n")
	return b.String()
}

// badTokenMessage is what an agent sees when its token does not authenticate.
// It never echoes the token.
func badTokenMessage(space core.SpaceRef, cause error) string {
	var b strings.Builder
	fmt.Fprintf(&b, "spec.sr.ht rejected this push.\n\n")
	fmt.Fprintf(&b, "  space: %s\n\n", space)
	fmt.Fprintf(&b, "  the agent token presented with this push was refused:\n")
	fmt.Fprintf(&b, "      %v\n\n", cause)
	fmt.Fprintf(&b, "Nothing was written. Agents write through the REST and MCP planes,\n")
	fmt.Fprintf(&b, "not over git; a token that works there is not a git credential.\n")
	return b.String()
}