sr/context: implement /contexts/{context}/... prefixed handlers

src/v/pandaproxy/schema_registry/BUILD

@@ -82,6 +82,7 @@ redpanda_cc_library(
         "avro.h",
         "compatibility.h",
         "configuration.h",
+        "context_router.h",
         "error.h",
         "errors.h",
         "exceptions.h",

src/v/pandaproxy/schema_registry/context_router.h

@@ -0,0 +1,49 @@
+// Copyright 2026 Redpanda Data, Inc.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.md
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0
+
+#pragma once
+
+#include "base/seastarx.h"
+#include "pandaproxy/schema_registry/errors.h"
+
+#include <seastar/core/sstring.hh>
+#include <seastar/http/request.hh>
+
+#include <fmt/format.h>
+
+#include <string_view>
+
+namespace pandaproxy::schema_registry {
+
+/// \brief Normalize a context name from a URL path parameter.
+inline ss::sstring normalize_context(std::string_view ctx) {
+    if (ctx.starts_with(':')) {
+        ctx.remove_prefix(1);
+    }
+
+    if (ctx.ends_with(':')) {
+        ctx.remove_suffix(1);
+    }
+
+    if (ctx.find(':') != std::string_view::npos) {
+        throw as_exception(context_invalid(ctx));
+    }
+
+    if (!ctx.starts_with('.')) {
+        return {fmt::format(".{}", ctx)};
+    }
+    return ss::sstring(ctx);
+}
+
+/// \brief Check if a string already has a context prefix.
+inline bool starts_with_context(std::string_view s) {
+    return s.starts_with(":.") || s.starts_with(":*:");
+}
+
+} // namespace pandaproxy::schema_registry

src/v/pandaproxy/schema_registry/error.cc

@@ -96,6 +96,8 @@ struct error_category final : std::error_category {
             return "The specified context is not empty";
         case error_code::subject_invalid:
             return "The specified subject is not valid";
+        case error_code::context_invalid:
+            return "The specified context name is not valid";
         }
         return "(unrecognized error)";
     }
@@ -164,6 +166,8 @@ struct error_category final : std::error_category {
             return reply_error_code::context_not_empty; // 42211
         case error_code::subject_invalid:
             return reply_error_code::subject_invalid; // 42208
+        case error_code::context_invalid:
+            return reply_error_code::bad_request; // 400
         }
         return {};
     }

src/v/pandaproxy/schema_registry/error.h

@@ -47,6 +47,7 @@ enum class error_code {
     writes_disabled,
     context_not_empty,
     subject_invalid,
+    context_invalid,
 };
 
 std::error_code make_error_code(error_code);

src/v/pandaproxy/schema_registry/errors.h

@@ -261,6 +261,12 @@ inline error_info context_not_empty(const context& ctx) {
       fmt::format("The specified context '{}' is not empty.", ctx())};
 }
 
+inline error_info context_invalid(std::string_view ctx) {
+    return error_info{
+      error_code::context_invalid,
+      fmt::format("The specified context '{}' is not valid.", ctx)};
+}
+
 inline bool failed_subject_schema_lookup(std::error_code ec) {
     return ec == error_code::subject_not_found
            || ec == error_code::subject_version_not_found;

src/v/pandaproxy/schema_registry/test/BUILD

@@ -348,6 +348,21 @@ redpanda_cc_gtest(
     ],
 )
 
+redpanda_cc_gtest(
+    name = "context_router_test",
+    timeout = "short",
+    srcs = [
+        "context_router.cc",
+    ],
+    deps = [
+        "//src/v/pandaproxy/schema_registry:core",
+        "//src/v/test_utils:gtest",
+        "@fmt",
+        "@googletest//:gtest",
+        "@seastar",
+    ],
+)
+
 redpanda_cc_btest(
     name = "test_json_schema",
     timeout = "short",

src/v/pandaproxy/schema_registry/test/context_router.cc

@@ -0,0 +1,68 @@
+// Copyright 2026 Redpanda Data, Inc.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.md
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0
+
+#include "pandaproxy/schema_registry/context_router.h"
+
+#include "pandaproxy/schema_registry/exceptions.h"
+
+#include <seastar/http/request.hh>
+
+#include <gtest/gtest.h>
+
+namespace pandaproxy::schema_registry {
+
+TEST(ContextRouterTest, StartsWithContextDefault) {
+    EXPECT_FALSE(starts_with_context(""));
+    EXPECT_FALSE(starts_with_context("my-topic"));
+    EXPECT_FALSE(starts_with_context("plain-subject"));
+}
+
+TEST(ContextRouterTest, StartsWithContextQualified) {
+    EXPECT_TRUE(starts_with_context(":.staging:my-topic"));
+    EXPECT_TRUE(starts_with_context(":.prod:"));
+    EXPECT_TRUE(starts_with_context(":.:my-topic"));
+}
+
+TEST(ContextRouterTest, StartsWithContextWildcard) {
+    EXPECT_TRUE(starts_with_context(":*:"));
+    EXPECT_TRUE(starts_with_context(":*:my-topic"));
+}
+
+TEST(ContextRouterTest, StartsWithContextEdgeCases) {
+    EXPECT_FALSE(starts_with_context(":"));
+    EXPECT_FALSE(starts_with_context(":foo"));
+    EXPECT_FALSE(starts_with_context(":*"));
+}
+
+TEST(ContextRouterTest, NormalizeContextWithDot) {
+    EXPECT_EQ(normalize_context(".staging"), ".staging");
+    EXPECT_EQ(normalize_context(".prod"), ".prod");
+    EXPECT_EQ(normalize_context("."), ".");
+}
+
+TEST(ContextRouterTest, NormalizeContextWithoutDot) {
+    EXPECT_EQ(normalize_context("staging"), ".staging");
+    EXPECT_EQ(normalize_context("prod"), ".prod");
+    EXPECT_EQ(normalize_context(""), ".");
+}
+
+TEST(ContextRouterTest, NormalizeContextStripColons) {
+    EXPECT_EQ(normalize_context(":.staging"), ".staging");
+    EXPECT_EQ(normalize_context(".staging:"), ".staging");
+    EXPECT_EQ(normalize_context(":.staging:"), ".staging");
+    EXPECT_EQ(normalize_context(":staging:"), ".staging");
+}
+
+TEST(ContextRouterTest, NormalizeContextRejectsEmbeddedColons) {
+    EXPECT_THROW(normalize_context(".:."), exception);
+    EXPECT_THROW(normalize_context("a:b"), exception);
+    EXPECT_THROW(normalize_context(":.a:b:"), exception);
+}
+
+} // namespace pandaproxy::schema_registry

src/v/utils/tests/BUILD

@@ -628,3 +628,16 @@ redpanda_cc_gtest(
         "@googletest//:gtest",
     ],
 )
+
+redpanda_cc_gtest(
+    name = "variant_test",
+    timeout = "short",
+    srcs = [
+        "variant_test.cc",
+    ],
+    deps = [
+        "//src/v/test_utils:gtest",
+        "//src/v/utils:variant",
+        "@googletest//:gtest",
+    ],
+)

src/v/utils/tests/variant_test.cc

@@ -0,0 +1,30 @@
+// Copyright 2026 Redpanda Data, Inc.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.md
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0
+
+#include "utils/variant.h"
+
+#include <gtest/gtest.h>
+
+#include <string>
+#include <type_traits>
+#include <variant>
+
+static_assert(std::is_same_v<
+              extend_variant_t<std::variant<int, double>, bool>,
+              std::variant<int, double, bool>>);
+
+static_assert(std::is_same_v<
+              extend_variant_t<std::variant<int>, double, std::string>,
+              std::variant<int, double, std::string>>);
+
+static_assert(std::is_same_v<
+              extend_variant_t<std::variant<int, double>>,
+              std::variant<int, double>>);
+
+TEST(VariantTest, CompileOnly) {}

src/v/utils/variant.h

@@ -35,3 +35,19 @@ using variant_of_identities = decltype(util::detail::variant_of_tags_impl(
 template<typename Variant>
 using tuple_of_identities = decltype(util::detail::tuple_of_tags_impl(
   std::type_identity<Variant>{}));
+
+namespace util::detail {
+
+template<typename Variant, typename... Extra>
+struct extend_variant;
+
+template<typename... Ts, typename... Extra>
+struct extend_variant<std::variant<Ts...>, Extra...> {
+    using type = std::variant<Ts..., Extra...>;
+};
+
+} // namespace util::detail
+
+template<typename Variant, typename... Extra>
+using extend_variant_t =
+  typename util::detail::extend_variant<Variant, Extra...>::type;