package remoteapi

import (
	"context"
	"database/sql"
	"errors"
	"fmt"

	remotesapi "github.com/dolthub/dolt/go/gen/proto/dolt/services/remotesapi/v1alpha1"
	"github.com/sirupsen/logrus"
	"github.com/vaughan0/go-ini"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"
	"sourcecraft.dev/bigbes/sr-ht-core/auth"
	"sourcecraft.dev/bigbes/sr-ht-core/config"
	"sourcecraft.dev/bigbes/sr-ht-core/database"

	"sourcecraft.dev/bigbes/sr-ht-dolt/authn"
	"sourcecraft.dev/bigbes/sr-ht-dolt/core"
	"sourcecraft.dev/bigbes/sr-ht-dolt/db"
	"sourcecraft.dev/bigbes/sr-ht-dolt/storage"
)

// Method-classification sets, copied verbatim from upstream remotesrv's
// interceptors.go (SUPER_USER_RPC_METHODS / CLONE_ADMIN_RPC_METHODS). Upstream
// drops the authenticated context and never sees the repo path, so it can only
// make a binary superuser/clone-admin decision; we keep its exact method lists
// but layer our own per-repo ACL check on top (see authorize).
//
//   - writeMethods are pushes: they require OpPush / AccessRW.
//   - readMethods are clone/pull/fetch reads: OpCloneRead / AccessRO.
//   - anything else is an unknown method and is denied (PermissionDenied),
//     matching upstream's "unknown rpc method" hard failure.
var (
	writeMethods = map[string]bool{
		"/dolt.services.remotesapi.v1alpha1.ChunkStoreService/AddTableFiles":      true,
		"/dolt.services.remotesapi.v1alpha1.ChunkStoreService/Commit":             true,
		"/dolt.services.remotesapi.v1alpha1.ChunkStoreService/GetUploadLocations": true,
	}
	readMethods = map[string]bool{
		"/dolt.services.remotesapi.v1alpha1.ChunkStoreService/GetDownloadLocations":    true,
		"/dolt.services.remotesapi.v1alpha1.ChunkStoreService/GetRepoMetadata":         true,
		"/dolt.services.remotesapi.v1alpha1.ChunkStoreService/HasChunks":               true,
		"/dolt.services.remotesapi.v1alpha1.ChunkStoreService/ListTableFiles":          true,
		"/dolt.services.remotesapi.v1alpha1.ChunkStoreService/RefreshTableFileUrl":     true,
		"/dolt.services.remotesapi.v1alpha1.ChunkStoreService/Root":                    true,
		"/dolt.services.remotesapi.v1alpha1.ChunkStoreService/StreamDownloadLocations": true,
		"/dolt.services.remotesapi.v1alpha1.ChunkStoreService/StreamChunkLocations":    true,
	}

	// rootMethod is the only read RPC the dolt client may send with no repo path
	// (a bare "what is the current root" ping used during dial/handshake). It is
	// treated as an unauthenticated-OK ping: it still authenticates the caller
	// (so a bad token is rejected) but skips the per-repo ACL check when no path
	// is present. Every real read carries a repo path and is checked normally.
	rootMethod = "/dolt.services.remotesapi.v1alpha1.ChunkStoreService/Root"
)

// repoRequest is the subset of every ChunkStoreService request message that
// carries the target repository, mirrored from upstream remotesrv's private
// repoRequest interface. All request types implement it.
type repoRequest interface {
	GetRepoId() *remotesapi.RepoId
	GetRepoPath() string
}

// repoStore is the narrow slice of db.Store the interceptor needs. Declaring it
// locally (rather than depending on *db.Store directly) keeps the authorization
// logic unit-testable with an in-memory stub — no Postgres required.
type repoStore interface {
	GetRepoByOwnerAndName(ctx context.Context, ownerUsername, name string) (*core.Repo, error)
	EffectiveAccess(ctx context.Context, userID, repoID int) (*core.AccessMode, error)
	// CreateRepo and DeleteRepo back push-to-create: an authenticated caller
	// touching a not-yet-existing repo in their own namespace has the row
	// inserted (CreateRepo) and, if the on-disk store then fails to materialize,
	// rolled back (DeleteRepo).
	CreateRepo(ctx context.Context, r *core.Repo) (*core.Repo, error)
	DeleteRepo(ctx context.Context, id int) error
}

// interceptor holds the collaborators the per-RPC auth/authz decision needs. It
// is installed on the remotesrv gRPC server via Options().
type interceptor struct {
	conf        ini.File
	service     string
	expectedAud string
	keys        authn.KeyStore
	logger      *logrus.Entry

	// pool is the shared database pool, threaded into the request context via
	// database.Context so db.FromContext-style lookups and the token resolvers
	// can reach it.
	pool *sql.DB

	// stores returns a repoStore for a request. Production binds a fresh
	// db.Store to the shared pool; tests inject a stub.
	stores func() repoStore

	// reposRoot is the absolute directory under which bare NBS stores live; it
	// resolves the on-disk path for a push-to-created repository.
	reposRoot string

	// createStore materializes the empty on-disk store for a push-to-created
	// repository. Production uses storage.InitEmptyStore; tests inject a fake.
	createStore func(ctx context.Context, absPath string) error
}

// newInterceptor builds the interceptor over the shared pool. It binds a fresh
// db.Store per request (the pool owns connection lifetime, ctx bounds each
// query). pool and keys must be non-nil.
func newInterceptor(conf ini.File, service, expectedAud string, pool *sql.DB, keys authn.KeyStore, reposRoot string, createStore func(ctx context.Context, absPath string) error, logger *logrus.Entry) *interceptor {
	if pool == nil {
		panic("remoteapi: newInterceptor requires a non-nil *sql.DB")
	}
	if logger == nil {
		logger = logrus.NewEntry(logrus.StandardLogger())
	}
	if createStore == nil {
		createStore = storage.InitEmptyStore
	}
	i := &interceptor{
		conf:        conf,
		service:     service,
		expectedAud: expectedAud,
		keys:        keys,
		logger:      logger,
		pool:        pool,
		reposRoot:   reposRoot,
		createStore: createStore,
	}
	i.stores = func() repoStore { return db.NewStore(pool) }
	return i
}

// dbHandle returns the shared pool threaded into request contexts.
func (i *interceptor) dbHandle() *sql.DB { return i.pool }

// Options returns the gRPC server options that install both interceptors,
// matching upstream ServerInterceptor.Options.
func (i *interceptor) Options() []grpc.ServerOption {
	return []grpc.ServerOption{
		grpc.ChainUnaryInterceptor(i.unary()),
		grpc.ChainStreamInterceptor(i.stream()),
	}
}

// classify maps a full gRPC method to its access op/mode. ok is false for an
// unknown method, which the caller denies.
func classify(fullMethod string) (op core.Op, mode core.AccessMode, ok bool) {
	if writeMethods[fullMethod] {
		return core.OpPush, core.AccessRW, true
	}
	if readMethods[fullMethod] {
		return core.OpCloneRead, core.AccessRO, true
	}
	return 0, "", false
}

// withServiceCtx augments the live per-RPC context (which carries the gRPC
// deadline, cancellation and incoming metadata) with the config and database
// values the authn resolvers and db.Store read from context. We augment the
// handler's context rather than starting from a stored base context so request
// deadlines and the incoming "authorization" metadata are preserved.
func (i *interceptor) withServiceCtx(ctx context.Context) context.Context {
	ctx = config.Context(ctx, i.conf, i.service)
	ctx = database.Context(ctx, i.dbHandle())
	return ctx
}

// authenticate resolves the caller from the incoming "authorization" metadata.
// It returns the caller (nil for an anonymous request) or a gRPC status error:
// Unauthenticated for a bad/forged/revoked credential, Unavailable for a
// transient backend failure (meta.sr.ht or the database unreachable).
func (i *interceptor) authenticate(ctx context.Context) (*auth.AuthContext, error) {
	header := ""
	if md, ok := metadata.FromIncomingContext(ctx); ok {
		if vals := md.Get("authorization"); len(vals) > 0 {
			header = vals[0]
		}
	}
	ac, err := authn.ResolveGRPCAuth(ctx, header, i.expectedAud, i.keys)
	if err != nil {
		if errors.Is(err, authn.ErrInvalidToken) {
			i.logger.Warnf("remotesapi authentication rejected: %v", err)
			return nil, status.Error(codes.Unauthenticated, "invalid or expired credentials")
		}
		// Transient: meta.sr.ht or the database is unreachable. Never surface as a
		// hard credential rejection — the client should retry.
		i.logger.Errorf("remotesapi authentication backend error: %v", err)
		return nil, status.Error(codes.Unavailable, "authentication temporarily unavailable")
	}
	return ac, nil
}

// authorize applies the grant gate and the per-repo ACL to an authenticated
// caller and returns a context carrying the resolved caller for the handler, or
// a gRPC status error. It is called once per unary request and once per stream
// message. The order mirrors the plan:
//
//  1. classify the method (unknown ⇒ PermissionDenied);
//  2. OAuth grant gate (TokenGrantsAllow) — PAT callers must carry
//     dolt.sr.ht/repos:RO for reads / :RW for pushes; anonymous, cookie and
//     dolt-key callers pass trivially;
//  3. extract the repo path (Root with no path ⇒ unauthenticated-OK ping);
//  4. load the repository row and the caller's effective ACL;
//  5. core.Allowed — on denial, NotFound for a PRIVATE repo the caller cannot
//     even see (no existence leak), PermissionDenied for a visible repo.
func (i *interceptor) authorize(ctx context.Context, ac *auth.AuthContext, fullMethod string, req any) (context.Context, error) {
	op, mode, ok := classify(fullMethod)
	if !ok {
		return nil, status.Errorf(codes.PermissionDenied, "unknown rpc method: %s", fullMethod)
	}

	// Grant gate is a global property of the token (independent of any repo), so
	// running it first leaks nothing about repository existence.
	if !authn.TokenGrantsAllow(ac, mode) {
		return nil, status.Errorf(codes.PermissionDenied,
			"token grants do not permit %s on %s repositories", mode, i.service)
	}

	rr, isRepoReq := req.(repoRequest)
	repoPath := ""
	if isRepoReq {
		repoPath = repoPathOf(rr)
	}
	if repoPath == "" {
		if fullMethod == rootMethod {
			// Handshake ping with no repo: authenticated but not repo-scoped.
			return authn.WithCaller(ctx, ac), nil
		}
		return nil, status.Error(codes.InvalidArgument, "request is missing a repository path")
	}

	owner, name, err := core.ParseRepoPath(repoPath)
	if err != nil {
		return nil, status.Errorf(codes.InvalidArgument, "invalid repository path %q: %v", repoPath, err)
	}

	caller := authn.AsCoreCaller(ac)

	store := i.stores()
	repo, err := store.GetRepoByOwnerAndName(ctx, owner, name)
	if err != nil {
		if errors.Is(err, db.ErrNotFound) {
			// Push-to-create: an authenticated, non-suspended caller touching a
			// not-yet-existing repo in THEIR OWN namespace (with a valid name)
			// has it transparently created — a PRIVATE row plus a genuinely
			// empty on-disk store — then proceeds through the normal ACL check
			// as the owner. Any other case (anonymous, suspended, another
			// user's namespace, invalid name) keeps returning NotFound, which
			// also avoids leaking existence of a stranger's private repos.
			if caller == nil || caller.Suspended || owner != caller.Username || core.ValidateName(name) != nil {
				return nil, status.Errorf(codes.NotFound, "repository %s/%s not found", owner, name)
			}
			repo, err = i.autoCreate(ctx, store, caller, owner, name)
			if err != nil {
				return nil, err
			}
		} else {
			i.logger.Errorf("remotesapi repo lookup %s/%s: %v", owner, name, err)
			return nil, status.Error(codes.Unavailable, "repository lookup temporarily unavailable")
		}
	}

	var aclMode *core.AccessMode
	if caller != nil {
		aclMode, err = store.EffectiveAccess(ctx, caller.UserID, repo.ID)
		if err != nil {
			i.logger.Errorf("remotesapi effective-access user=%d repo=%d: %v", caller.UserID, repo.ID, err)
			return nil, status.Error(codes.Unavailable, "authorization temporarily unavailable")
		}
	}

	if !core.Allowed(caller, repo, aclMode, op) {
		if core.NotFoundForPrivate(caller, repo, aclMode) {
			return nil, status.Errorf(codes.NotFound, "repository %s/%s not found", owner, name)
		}
		return nil, status.Errorf(codes.PermissionDenied, "%s denied on %s/%s", op, owner, name)
	}

	return authn.WithCaller(ctx, ac), nil
}

// autoCreate transparently creates a PRIVATE repository owned by caller under
// owner/name (push-to-create). It inserts the row and materializes a genuinely
// empty on-disk store. A concurrent first-push that already created the row is
// tolerated: the caller loses the CreateRepo race, re-fetches the winner's row,
// and does NOT re-create the store. If the store fails to materialize, the row
// is rolled back so a later push retries cleanly. Preconditions (authenticated,
// non-suspended, owns the namespace, valid name) are checked by the caller.
func (i *interceptor) autoCreate(ctx context.Context, store repoStore, caller *core.Caller, owner, name string) (*core.Repo, error) {
	newRepo := &core.Repo{
		Name:       name,
		OwnerID:    caller.UserID,
		OwnerName:  owner,
		Path:       storage.RepoDiskPath(i.reposRoot, owner, name),
		Visibility: core.VisibilityPrivate,
	}
	created, cerr := store.CreateRepo(ctx, newRepo)
	if cerr != nil {
		if errors.Is(cerr, db.ErrNameTaken) {
			// Lost the race: another first-push created the row (and its store).
			// Adopt the winner's row; do not touch disk.
			repo, gerr := store.GetRepoByOwnerAndName(ctx, owner, name)
			if gerr != nil {
				i.logger.Errorf("remotesapi auto-create refetch %s/%s: %v", owner, name, gerr)
				return nil, status.Error(codes.Unavailable, "repository creation temporarily unavailable")
			}
			return repo, nil
		}
		i.logger.Errorf("remotesapi auto-create %s/%s: %v", owner, name, cerr)
		return nil, status.Error(codes.Unavailable, "repository creation temporarily unavailable")
	}

	if serr := i.createStore(ctx, created.Path); serr != nil {
		// The row exists but the store does not: roll the row back (best effort)
		// so the repo does not linger half-created and a retry can succeed.
		if derr := store.DeleteRepo(ctx, created.ID); derr != nil {
			i.logger.Errorf("remotesapi auto-create rollback %s/%s (id=%d): %v", owner, name, created.ID, derr)
		}
		i.logger.Errorf("remotesapi auto-create store %s/%s: %v", owner, name, serr)
		return nil, status.Error(codes.Unavailable, "repository creation temporarily unavailable")
	}
	return created, nil
}

// repoPathOf extracts the target repo path from a request without panicking on
// an absent path (upstream's getRepoPath panics on empty path + nil repo id).
// It returns "" when neither is present so the caller can treat Root specially.
func repoPathOf(req repoRequest) string {
	if p := req.GetRepoPath(); p != "" {
		return p
	}
	if id := req.GetRepoId(); id != nil {
		return fmt.Sprintf("%s/%s", id.Org, id.RepoName)
	}
	return ""
}

// unary is the unary server interceptor: authenticate once, authorize the
// request, then invoke the handler with the caller-carrying context.
func (i *interceptor) unary() grpc.UnaryServerInterceptor {
	return func(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
		ctx = i.withServiceCtx(ctx)
		ac, err := i.authenticate(ctx)
		if err != nil {
			return nil, err
		}
		authedCtx, err := i.authorize(ctx, ac, info.FullMethod, req)
		if err != nil {
			return nil, err
		}
		return handler(authedCtx, req)
	}
}

// stream is the stream server interceptor. Authentication happens once at
// stream start; the repo-path authorization is re-checked for every message the
// client sends (the streaming reads carry a repo path per message), matching the
// plan's "override Context() and check each RecvMsg" design.
func (i *interceptor) stream() grpc.StreamServerInterceptor {
	return func(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
		ctx := i.withServiceCtx(ss.Context())
		ac, err := i.authenticate(ctx)
		if err != nil {
			return err
		}
		// A stream to an unknown method is denied before the handler runs.
		if _, _, ok := classify(info.FullMethod); !ok {
			return status.Errorf(codes.PermissionDenied, "unknown rpc method: %s", info.FullMethod)
		}
		wrapped := &authStream{
			ServerStream: ss,
			ctx:          authn.WithCaller(ctx, ac),
			i:            i,
			ac:           ac,
			fullMethod:   info.FullMethod,
		}
		return handler(srv, wrapped)
	}
}

// authStream wraps a grpc.ServerStream so the handler sees the caller-carrying
// context and every received message is re-authorized against its repo path.
type authStream struct {
	grpc.ServerStream
	ctx        context.Context
	i          *interceptor
	ac         *auth.AuthContext
	fullMethod string
}

// Context returns the authenticated, caller-carrying context.
func (s *authStream) Context() context.Context { return s.ctx }

// RecvMsg receives the next message and re-authorizes it against its repo path
// before handing it to the handler. An authorization failure aborts the stream.
func (s *authStream) RecvMsg(m any) error {
	if err := s.ServerStream.RecvMsg(m); err != nil {
		return err
	}
	if _, err := s.i.authorize(s.ctx, s.ac, s.fullMethod, m); err != nil {
		return err
	}
	return nil
}