admin: add read-only DynamoDB tables endpoints (P1)

GET /admin/api/v1/dynamo/tables and GET /admin/api/v1/dynamo/tables/{name}
land the design doc Section 4.1 read-only paths. Both go through the
same protect chain (BodyLimit -> SessionAuth -> Audit -> CSRF) as
write endpoints; Audit is a no-op on GET so dashboard polling does not
flood the audit log.

Adapter side (SigV4 bypass per Section 3.2):
- adapter.AdminListTables / AdminDescribeTable internal entrypoints
- AdminTableSummary DTO keeps dynamoTableSchema private; admin gets a
  stable struct that does not drift with the wire format

Admin handler (internal/admin/dynamo_handler.go):
- TablesSource interface (production wired, tests stub)
- limit (default 100, hard max 1000) + opaque base64 next_token
  pagination matching design Section 4.3
- vanished-cursor fast-forward so a deleted-between-pages name does
  not stall the SPA
- internal errors hidden behind dynamo_list_failed / dynamo_describe_failed
  codes; raw err.Error() is never leaked to clients

Server wiring:
- ServerDeps.Tables is optional; nil leaves the dynamo paths off the
  wire entirely (returns the standard unknown_endpoint 404)
- main_admin.go bridges *adapter.DynamoDBServer via dynamoTablesBridge,
  translating adapter.AdminTableSummary <-> admin.DynamoTableSummary
  (isomorphic so any field drift breaks the build)

Author: bootjp

adapter/dynamodb_admin.go

@@ -0,0 +1,84 @@
+package adapter
+
+import (
+	"context"
+	"sort"
+)
+
+// AdminTableSummary is the table-level information the admin dashboard
+// surfaces for a single Dynamo-compatible table. It deliberately
+// projects only the fields the dashboard needs so the package's
+// wire-format types (dynamoTableSchema and friends) stay internal.
+type AdminTableSummary struct {
+	Name                   string
+	PartitionKey           string
+	SortKey                string
+	Generation             uint64
+	GlobalSecondaryIndexes []AdminGSISummary
+}
+
+// AdminGSISummary mirrors AdminTableSummary for a single GSI.
+type AdminGSISummary struct {
+	Name           string
+	PartitionKey   string
+	SortKey        string
+	ProjectionType string
+}
+
+// AdminListTables returns every Dynamo-style table this server knows
+// about, in the lexicographic order the metadata index produces.
+// Intended for the in-process admin listener as the SigV4-free
+// counterpart to the listTables HTTP handler; both share the same
+// underlying lookup so the two views cannot drift.
+func (d *DynamoDBServer) AdminListTables(ctx context.Context) ([]string, error) {
+	return d.listTableNames(ctx)
+}
+
+// AdminDescribeTable returns a schema snapshot for name. The triple
+// (result, present, error) lets admin callers distinguish a genuine
+// "not found" from a storage error without sniffing sentinels: when
+// the table is missing the function returns (nil, false, nil).
+func (d *DynamoDBServer) AdminDescribeTable(ctx context.Context, name string) (*AdminTableSummary, bool, error) {
+	if err := d.ensureLegacyTableMigration(ctx, name); err != nil {
+		return nil, false, err
+	}
+	schema, exists, err := d.loadTableSchema(ctx, name)
+	if err != nil {
+		return nil, false, err
+	}
+	if !exists {
+		return nil, false, nil
+	}
+	return summaryFromSchema(schema), true, nil
+}
+
+func summaryFromSchema(s *dynamoTableSchema) *AdminTableSummary {
+	out := &AdminTableSummary{
+		Name:         s.TableName,
+		PartitionKey: s.PrimaryKey.HashKey,
+		SortKey:      s.PrimaryKey.RangeKey,
+		Generation:   s.Generation,
+	}
+	if len(s.GlobalSecondaryIndexes) == 0 {
+		return out
+	}
+	names := make([]string, 0, len(s.GlobalSecondaryIndexes))
+	for n := range s.GlobalSecondaryIndexes {
+		names = append(names, n)
+	}
+	// Sort so the JSON the admin handler emits is deterministic; map
+	// iteration order would otherwise produce an unstable output that
+	// breaks both UI diffing and snapshot tests.
+	sort.Strings(names)
+	out.GlobalSecondaryIndexes = make([]AdminGSISummary, 0, len(names))
+	for _, name := range names {
+		gsi := s.GlobalSecondaryIndexes[name]
+		out.GlobalSecondaryIndexes = append(out.GlobalSecondaryIndexes, AdminGSISummary{
+			Name:           name,
+			PartitionKey:   gsi.KeySchema.HashKey,
+			SortKey:        gsi.KeySchema.RangeKey,
+			ProjectionType: gsi.Projection.ProjectionType,
+		})
+	}
+	return out
+}

adapter/dynamodb_admin_test.go

@@ -0,0 +1,183 @@
+package adapter
+
+import (
+	"context"
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
+	ddbTypes "github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
+	"github.com/stretchr/testify/require"
+)
+
+// TestDynamoDB_AdminListTables_Empty exercises the SigV4-bypass admin
+// entrypoint on a server that has no Dynamo tables. The expected shape
+// is an empty (non-nil) slice so the admin JSON response stays a valid
+// array rather than `null`, matching the design doc 4.3 contract.
+func TestDynamoDB_AdminListTables_Empty(t *testing.T) {
+	t.Parallel()
+	nodes, _, _ := createNode(t, 1)
+	defer shutdown(nodes)
+
+	got, err := nodes[0].dynamoServer.AdminListTables(context.Background())
+	require.NoError(t, err)
+	require.Empty(t, got)
+}
+
+// TestDynamoDB_AdminListTables_Sorted verifies that the admin entrypoint
+// returns table names in lexicographic order, matching the listTables
+// HTTP handler so the two admin views (SigV4 and bypass) cannot drift.
+func TestDynamoDB_AdminListTables_Sorted(t *testing.T) {
+	t.Parallel()
+	nodes, _, _ := createNode(t, 1)
+	defer shutdown(nodes)
+
+	client := newDynamoClient(t, nodes[0].dynamoAddress)
+	ctx := context.Background()
+
+	for _, name := range []string{"zeta", "alpha", "mu"} {
+		_, err := client.CreateTable(ctx, &dynamodb.CreateTableInput{
+			TableName:   aws.String(name),
+			BillingMode: ddbTypes.BillingModePayPerRequest,
+			AttributeDefinitions: []ddbTypes.AttributeDefinition{
+				{AttributeName: aws.String("pk"), AttributeType: ddbTypes.ScalarAttributeTypeS},
+			},
+			KeySchema: []ddbTypes.KeySchemaElement{
+				{AttributeName: aws.String("pk"), KeyType: ddbTypes.KeyTypeHash},
+			},
+		})
+		require.NoError(t, err)
+	}
+
+	got, err := nodes[0].dynamoServer.AdminListTables(ctx)
+	require.NoError(t, err)
+	require.Equal(t, []string{"alpha", "mu", "zeta"}, got)
+}
+
+// TestDynamoDB_AdminDescribeTable_Missing checks the (nil, false, nil)
+// "not found" contract — admin callers must be able to tell a missing
+// table apart from a storage error without sniffing sentinels.
+func TestDynamoDB_AdminDescribeTable_Missing(t *testing.T) {
+	t.Parallel()
+	nodes, _, _ := createNode(t, 1)
+	defer shutdown(nodes)
+
+	summary, exists, err := nodes[0].dynamoServer.AdminDescribeTable(context.Background(), "absent")
+	require.NoError(t, err)
+	require.False(t, exists)
+	require.Nil(t, summary)
+}
+
+// TestDynamoDB_AdminDescribeTable_Composite covers the simple-key happy
+// path: a table with hash + range key and no GSIs. The admin summary
+// must mirror the schema's primary key fields exactly.
+func TestDynamoDB_AdminDescribeTable_Composite(t *testing.T) {
+	t.Parallel()
+	nodes, _, _ := createNode(t, 1)
+	defer shutdown(nodes)
+
+	client := newDynamoClient(t, nodes[0].dynamoAddress)
+	ctx := context.Background()
+
+	_, err := client.CreateTable(ctx, &dynamodb.CreateTableInput{
+		TableName:   aws.String("orders"),
+		BillingMode: ddbTypes.BillingModePayPerRequest,
+		AttributeDefinitions: []ddbTypes.AttributeDefinition{
+			{AttributeName: aws.String("customer"), AttributeType: ddbTypes.ScalarAttributeTypeS},
+			{AttributeName: aws.String("orderID"), AttributeType: ddbTypes.ScalarAttributeTypeS},
+		},
+		KeySchema: []ddbTypes.KeySchemaElement{
+			{AttributeName: aws.String("customer"), KeyType: ddbTypes.KeyTypeHash},
+			{AttributeName: aws.String("orderID"), KeyType: ddbTypes.KeyTypeRange},
+		},
+	})
+	require.NoError(t, err)
+
+	summary, exists, err := nodes[0].dynamoServer.AdminDescribeTable(ctx, "orders")
+	require.NoError(t, err)
+	require.True(t, exists)
+	require.NotNil(t, summary)
+	require.Equal(t, "orders", summary.Name)
+	require.Equal(t, "customer", summary.PartitionKey)
+	require.Equal(t, "orderID", summary.SortKey)
+	require.NotZero(t, summary.Generation)
+	require.Empty(t, summary.GlobalSecondaryIndexes)
+}
+
+// TestDynamoDB_AdminDescribeTable_GSI_SortedDeterministic exercises the
+// GSI projection path. Two indexes are added in deliberately reversed
+// alphabetical order to confirm summaryFromSchema's Sort.Strings call —
+// without it, map iteration order would produce a flaky output.
+func TestDynamoDB_AdminDescribeTable_GSI_SortedDeterministic(t *testing.T) {
+	t.Parallel()
+	nodes, _, _ := createNode(t, 1)
+	defer shutdown(nodes)
+
+	client := newDynamoClient(t, nodes[0].dynamoAddress)
+	ctx := context.Background()
+
+	_, err := client.CreateTable(ctx, &dynamodb.CreateTableInput{
+		TableName:   aws.String("threads"),
+		BillingMode: ddbTypes.BillingModePayPerRequest,
+		AttributeDefinitions: []ddbTypes.AttributeDefinition{
+			{AttributeName: aws.String("threadId"), AttributeType: ddbTypes.ScalarAttributeTypeS},
+			{AttributeName: aws.String("status"), AttributeType: ddbTypes.ScalarAttributeTypeS},
+			{AttributeName: aws.String("owner"), AttributeType: ddbTypes.ScalarAttributeTypeS},
+			{AttributeName: aws.String("createdAt"), AttributeType: ddbTypes.ScalarAttributeTypeS},
+		},
+		KeySchema: []ddbTypes.KeySchemaElement{
+			{AttributeName: aws.String("threadId"), KeyType: ddbTypes.KeyTypeHash},
+		},
+		GlobalSecondaryIndexes: []ddbTypes.GlobalSecondaryIndex{
+			{
+				IndexName: aws.String("zStatusIndex"),
+				KeySchema: []ddbTypes.KeySchemaElement{
+					{AttributeName: aws.String("status"), KeyType: ddbTypes.KeyTypeHash},
+					{AttributeName: aws.String("createdAt"), KeyType: ddbTypes.KeyTypeRange},
+				},
+				Projection: &ddbTypes.Projection{ProjectionType: ddbTypes.ProjectionTypeAll},
+			},
+			{
+				IndexName: aws.String("aOwnerIndex"),
+				KeySchema: []ddbTypes.KeySchemaElement{
+					{AttributeName: aws.String("owner"), KeyType: ddbTypes.KeyTypeHash},
+				},
+				Projection: &ddbTypes.Projection{ProjectionType: ddbTypes.ProjectionTypeKeysOnly},
+			},
+		},
+	})
+	require.NoError(t, err)
+
+	summary, exists, err := nodes[0].dynamoServer.AdminDescribeTable(ctx, "threads")
+	require.NoError(t, err)
+	require.True(t, exists)
+	require.NotNil(t, summary)
+	require.Equal(t, "threadId", summary.PartitionKey)
+	require.Empty(t, summary.SortKey)
+
+	require.Len(t, summary.GlobalSecondaryIndexes, 2)
+	// Names sorted lexicographically: "aOwnerIndex" < "zStatusIndex".
+	require.Equal(t, "aOwnerIndex", summary.GlobalSecondaryIndexes[0].Name)
+	require.Equal(t, "owner", summary.GlobalSecondaryIndexes[0].PartitionKey)
+	require.Empty(t, summary.GlobalSecondaryIndexes[0].SortKey)
+	require.Equal(t, string(ddbTypes.ProjectionTypeKeysOnly), summary.GlobalSecondaryIndexes[0].ProjectionType)
+
+	require.Equal(t, "zStatusIndex", summary.GlobalSecondaryIndexes[1].Name)
+	require.Equal(t, "status", summary.GlobalSecondaryIndexes[1].PartitionKey)
+	require.Equal(t, "createdAt", summary.GlobalSecondaryIndexes[1].SortKey)
+	require.Equal(t, string(ddbTypes.ProjectionTypeAll), summary.GlobalSecondaryIndexes[1].ProjectionType)
+}
+
+func newDynamoClient(t *testing.T, address string) *dynamodb.Client {
+	t.Helper()
+	cfg, err := config.LoadDefaultConfig(context.Background(),
+		config.WithRegion("us-west-2"),
+		config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider("dummy", "dummy", "")),
+	)
+	require.NoError(t, err)
+	return dynamodb.NewFromConfig(cfg, func(o *dynamodb.Options) {
+		o.BaseEndpoint = aws.String("http://" + address)
+	})
+}