feat: add IBigSegmentStore interface + Redis and DynamoDB stores (#536)

Ticket: SDK-2363 · Follows #534

## Summary

Adds the public `IBigSegmentStore` interface, the `Membership` /
`StoreMetadata` value types, and concrete DynamoDB + Redis
implementations. Schema strings match what the Relay Proxy writes.

```cpp
auto redis_store = RedisBigSegmentStore::Create("redis://localhost:6379", "prefix");
auto dynamo_store = DynamoDBBigSegmentStore::Create("my-table", "prefix", options);
```

## Design notes

- **`synchronizedOn` parsing.** Stored as DynamoDB N / Redis string.
Both stores reject malformed values (non-numeric strings, wrong DynamoDB
attribute type) rather than silently returning 0, matching the existing
`dynamodb_source.cpp` row validation.
- **No hashing here.** The interface contract says callers pass the
already-hashed base64 SHA-256 context key; the SDK will hash in the
wrapper, not the store implementations.

## Not in scope

- `BigSegmentsBuilder` config plumbing.
- `BigSegmentStoreWrapper` (LRU cache + staleness polling) and the
hashing path.
- Evaluator wiring / replacing the `rules.cpp` big-segments TODO.

## Test plan

- [x] 7 `Membership` unit tests pass.
- [x] 9 `RedisBigSegmentStore` integration tests pass against Redis 7
(docker).
- [x] 8 `DynamoDBBigSegmentStore` integration tests pass against
DynamoDB Local.
- [x] Full server-sdk test suite (468 tests) still green.
- [x] Schema constants match what Relay writes (verified against the
LocalStack/Redis fixture data).

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Medium Risk**
> New external-store read path for segment targeting with strict schema
validation; not yet wired into evaluation, so production flag behavior
is unchanged until follow-up PRs land.
> 
> **Overview**
> Introduces a **Big Segments persistent store** surface for the server
SDK: public `IBigSegmentStore`, inline `Membership` / `StoreMetadata`,
and read-only **Redis** and **DynamoDB** backends that follow Relay’s
key/schema layout (prefix scoping, include/exclude refs, sync metadata).
> 
> Stores perform point lookups by opaque context hash, build membership
via `Membership::FromSegmentRefs` (inclusion wins on overlap), and
return errors on malformed DynamoDB attribute types or non-numeric sync
timestamps instead of silent empty/zero results. **Config, caching
wrapper, hashing, and flag evaluation wiring are not included** in this
change.
> 
> Integration tests cover membership/metadata behavior, prefix
isolation, and error paths against Redis and DynamoDB Local.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
d2622d9. Bugbot is set up for automated
code reviews on this repo. Configure
[here](https://www.cursor.com/dashboard/bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: beekld

libs/server-sdk-dynamodb-source/include/launchdarkly/server_side/integrations/dynamodb/dynamodb_big_segment_store.hpp

@@ -0,0 +1,82 @@
+/** @file dynamodb_big_segment_store.hpp
+ * @brief Server-Side DynamoDB Big Segments Store
+ */
+
+#pragma once
+
+#include <launchdarkly/server_side/integrations/big_segments/ibig_segment_store.hpp>
+#include <launchdarkly/server_side/integrations/dynamodb/options.hpp>
+
+#include <tl/expected.hpp>
+
+#include <memory>
+#include <string>
+
+namespace Aws::DynamoDB {
+class DynamoDBClient;
+}
+
+namespace launchdarkly::server_side::integrations {
+
+/**
+ * @brief DynamoDBBigSegmentStore is a Big Segments persistent store backed by
+ * Amazon DynamoDB.
+ *
+ * Call DynamoDBBigSegmentStore::Create to obtain a new instance, then pass it
+ * to the SDK via the Big Segments config builder.
+ *
+ * The DynamoDB table must already exist and follow the LaunchDarkly schema:
+ * a String partition key named `namespace` and a String sort key named `key`.
+ * The same table can be shared with @ref DynamoDBDataSource — Big Segments
+ * rows occupy their own partition-key values and do not conflict with
+ * flag/segment rows. The LaunchDarkly Relay Proxy is responsible for
+ * populating Big Segments data in this table; this class only reads from it.
+ *
+ * This implementation is backed by the AWS SDK for C++.
+ */
+class DynamoDBBigSegmentStore final : public IBigSegmentStore {
+   public:
+    /**
+     * @brief Creates a new DynamoDBBigSegmentStore, or returns an error if
+     * construction failed.
+     *
+     * @param table_name Name of the DynamoDB table to read from. The table
+     * must already exist; this class does not create it.
+     *
+     * @param prefix Optional namespace prefix. When non-empty, Big Segments
+     * rows live under partition keys `<prefix>:big_segments_user` and
+     * `<prefix>:big_segments_metadata`. This allows multiple LaunchDarkly
+     * environments to share a single table.
+     *
+     * @param options Optional AWS DynamoDB client configuration. See
+     * @ref DynamoDBClientOptions. When defaulted, the AWS SDK resolves
+     * region, endpoint, and credentials from the standard provider chain
+     * (environment variables, shared config files, instance metadata).
+     *
+     * @return A DynamoDBBigSegmentStore, or an error if construction failed.
+     */
+    static tl::expected<std::unique_ptr<DynamoDBBigSegmentStore>, std::string>
+    Create(std::string table_name,
+           std::string prefix,
+           DynamoDBClientOptions options = {});
+
+    [[nodiscard]] GetMembershipResult GetMembership(
+        std::string const& context_hash) const override;
+    [[nodiscard]] GetMetadataResult GetMetadata() const override;
+
+    ~DynamoDBBigSegmentStore() override;
+
+   private:
+    DynamoDBBigSegmentStore(
+        std::unique_ptr<Aws::DynamoDB::DynamoDBClient> client,
+        std::string table_name,
+        std::string prefix);
+
+    std::unique_ptr<Aws::DynamoDB::DynamoDBClient> client_;
+    std::string const table_name_;
+    std::string const prefix_;
+    std::string const user_namespace_;
+    std::string const metadata_namespace_;
+};
+
+}  // namespace launchdarkly::server_side::integrations

libs/server-sdk-dynamodb-source/src/CMakeLists.txt

@@ -14,6 +14,7 @@ target_sources(${LIBNAME}
         PRIVATE
         ${HEADER_LIST}
         dynamodb_source.cpp
+        dynamodb_big_segment_store.cpp
         aws_sdk_guard.cpp
         client_factory.cpp
 )

libs/server-sdk-dynamodb-source/src/dynamodb_attributes.hpp

@@ -18,4 +18,16 @@ inline constexpr char kItemAttribute[] = "item";
 //   {namespace: "myprefix:$inited", key: "myprefix:$inited"}.
 inline constexpr char kInitedNamespace[] = "$inited";
 
+// Big Segments schema. Membership rows use partition key
+// "{prefix}:big_segments_user" and sort key {context_hash}, with
+// "included" / "excluded" String Set attributes naming segment refs. The
+// metadata row uses partition key AND sort key both set to
+// "{prefix}:big_segments_metadata", with the sync timestamp stored as a
+// Number under "synchronizedOn".
+inline constexpr char kBigSegmentsUserNamespace[] = "big_segments_user";
+inline constexpr char kBigSegmentsMetadataNamespace[] = "big_segments_metadata";
+inline constexpr char kBigSegmentsIncludedAttribute[] = "included";
+inline constexpr char kBigSegmentsExcludedAttribute[] = "excluded";
+inline constexpr char kBigSegmentsSyncTimeAttribute[] = "synchronizedOn";
+
 }  // namespace launchdarkly::server_side::integrations::detail

libs/server-sdk-dynamodb-source/src/dynamodb_big_segment_store.cpp

@@ -0,0 +1,163 @@
+#include <launchdarkly/server_side/integrations/dynamodb/dynamodb_big_segment_store.hpp>
+
+#include "aws_sdk_guard.hpp"
+#include "client_factory.hpp"
+#include "dynamodb_attributes.hpp"
+#include "prefix.hpp"
+
+#include <aws/core/utils/Outcome.h>
+#include <aws/dynamodb/DynamoDBClient.h>
+#include <aws/dynamodb/model/AttributeValue.h>
+#include <aws/dynamodb/model/GetItemRequest.h>
+
+#include <cerrno>
+#include <cstdint>
+#include <cstdlib>
+#include <exception>
+#include <utility>
+
+namespace launchdarkly::server_side::integrations {
+
+namespace {
+
+using detail::kBigSegmentsExcludedAttribute;
+using detail::kBigSegmentsIncludedAttribute;
+using detail::kBigSegmentsMetadataNamespace;
+using detail::kBigSegmentsSyncTimeAttribute;
+using detail::kBigSegmentsUserNamespace;
+using detail::kPartitionKey;
+using detail::kSortKey;
+using detail::PrefixedNamespace;
+
+}  // namespace
+
+tl::expected<std::unique_ptr<DynamoDBBigSegmentStore>, std::string>
+DynamoDBBigSegmentStore::Create(std::string table_name,
+                                std::string prefix,
+                                DynamoDBClientOptions options) {
+    try {
+        detail::AwsSdkGuard::Ensure();
+        auto maybe_client = detail::BuildDynamoDBClient(options);
+        if (!maybe_client) {
+            return tl::make_unexpected(std::move(maybe_client.error()));
+        }
+        return std::unique_ptr<DynamoDBBigSegmentStore>(
+            new DynamoDBBigSegmentStore(std::move(*maybe_client),
+                                        std::move(table_name),
+                                        std::move(prefix)));
+    } catch (std::exception const& e) {
+        return tl::make_unexpected(e.what());
+    }
+}
+
+DynamoDBBigSegmentStore::DynamoDBBigSegmentStore(
+    std::unique_ptr<Aws::DynamoDB::DynamoDBClient> client,
+    std::string table_name,
+    std::string prefix)
+    : client_(std::move(client)),
+      table_name_(std::move(table_name)),
+      prefix_(std::move(prefix)),
+      user_namespace_(PrefixedNamespace(prefix_, kBigSegmentsUserNamespace)),
+      metadata_namespace_(
+          PrefixedNamespace(prefix_, kBigSegmentsMetadataNamespace)) {}
+
+DynamoDBBigSegmentStore::~DynamoDBBigSegmentStore() = default;
+
+IBigSegmentStore::GetMembershipResult DynamoDBBigSegmentStore::GetMembership(
+    std::string const& context_hash) const {
+    Aws::DynamoDB::Model::GetItemRequest request;
+    request.SetTableName(table_name_);
+    request.SetConsistentRead(true);
+    request.AddKey(kPartitionKey,
+                   Aws::DynamoDB::Model::AttributeValue{user_namespace_});
+    request.AddKey(kSortKey,
+                   Aws::DynamoDB::Model::AttributeValue{context_hash});
+
+    auto outcome = client_->GetItem(request);
+    if (!outcome.IsSuccess()) {
+        return tl::make_unexpected(outcome.GetError().GetMessage());
+    }
+
+    auto const& item = outcome.GetResult().GetItem();
+
+    std::vector<std::string> included;
+    std::vector<std::string> excluded;
+
+    // GetSS() silently returns an empty vector if the attribute is not
+    // actually a String Set, so check the type explicitly before reading.
+    if (auto const it = item.find(kBigSegmentsIncludedAttribute);
+        it != item.end()) {
+        if (it->second.GetType() !=
+            Aws::DynamoDB::Model::ValueType::STRING_SET) {
+            return tl::make_unexpected(
+                std::string("DynamoDB Big Segments '") +
+                kBigSegmentsIncludedAttribute +
+                "' is not of type STRING_SET");
+        }
+        for (auto const& ref : it->second.GetSS()) {
+            included.emplace_back(ref);
+        }
+    }
+    if (auto const it = item.find(kBigSegmentsExcludedAttribute);
+        it != item.end()) {
+        if (it->second.GetType() !=
+            Aws::DynamoDB::Model::ValueType::STRING_SET) {
+            return tl::make_unexpected(
+                std::string("DynamoDB Big Segments '") +
+                kBigSegmentsExcludedAttribute +
+                "' is not of type STRING_SET");
+        }
+        for (auto const& ref : it->second.GetSS()) {
+            excluded.emplace_back(ref);
+        }
+    }
+
+    return Membership::FromSegmentRefs(included, excluded);
+}
+
+IBigSegmentStore::GetMetadataResult DynamoDBBigSegmentStore::GetMetadata()
+    const {
+    Aws::DynamoDB::Model::GetItemRequest request;
+    request.SetTableName(table_name_);
+    request.SetConsistentRead(true);
+    request.AddKey(kPartitionKey,
+                   Aws::DynamoDB::Model::AttributeValue{metadata_namespace_});
+    request.AddKey(kSortKey,
+                   Aws::DynamoDB::Model::AttributeValue{metadata_namespace_});
+
+    auto outcome = client_->GetItem(request);
+    if (!outcome.IsSuccess()) {
+        return tl::make_unexpected(outcome.GetError().GetMessage());
+    }
+
+    auto const& item = outcome.GetResult().GetItem();
+    if (item.empty()) {
+        return std::nullopt;
+    }
+
+    auto const it = item.find(kBigSegmentsSyncTimeAttribute);
+    if (it == item.end()) {
+        // "absent" sync time is treated as never synchronized rather than
+        // an error; the wrapper marks the store stale based on the
+        // resulting nullopt.
+        return std::nullopt;
+    }
+
+    auto const& raw = it->second.GetN();
+    if (raw.empty()) {
+        return tl::make_unexpected(
+            "DynamoDB Big Segments 'synchronizedOn' is empty or not type N");
+    }
+
+    errno = 0;
+    char* end = nullptr;
+    long long const parsed = std::strtoll(raw.c_str(), &end, 10);
+    if (errno != 0 || end == raw.c_str() || *end != '\0') {
+        return tl::make_unexpected(
+            "DynamoDB Big Segments 'synchronizedOn' is not a valid integer");
+    }
+
+    return StoreMetadata{std::chrono::milliseconds{parsed}};
+}
+
+}  // namespace launchdarkly::server_side::integrations