feat: Add typed DynamoDB query, write, and batch APIs with auto-pagination

src/Clients/Goa.Clients.Dynamo.Generator/Attributes/DynamoConverterAttributeHandler.cs

@@ -0,0 +1,44 @@
+using Microsoft.CodeAnalysis;
+using Goa.Clients.Dynamo.Generator.Models;
+
+namespace Goa.Clients.Dynamo.Generator.Attributes;
+
+/// <summary>
+/// Handles the DynamoConverterAttribute.
+/// </summary>
+public class DynamoConverterAttributeHandler : IAttributeHandler
+{
+    public string AttributeTypeName => "Goa.Clients.Dynamo.DynamoConverterAttribute";
+
+    public bool CanHandle(AttributeData attributeData)
+    {
+        return attributeData.AttributeClass?.ToDisplayString() == AttributeTypeName;
+    }
+
+    public AttributeInfo? ParseAttribute(AttributeData attributeData)
+    {
+        if (!CanHandle(attributeData))
+        {
+            return null;
+        }
+
+        // Extract converter type from constructor arg
+        if (attributeData.ConstructorArguments.Length > 0 &&
+            attributeData.ConstructorArguments[0].Value is INamedTypeSymbol converterType)
+        {
+            return new DynamoConverterAttributeInfo
+            {
+                AttributeData = attributeData,
+                AttributeTypeName = AttributeTypeName,
+                ConverterTypeName = converterType.ToDisplayString()
+            };
+        }
+
+        return null;
+    }
+
+    public void ValidateAttribute(AttributeInfo attributeInfo, ISymbol symbol, Action<Diagnostic> reportDiagnostic)
+    {
+        // No additional validation needed at this stage
+    }
+}

src/Clients/Goa.Clients.Dynamo.Generator/CodeGeneration/ReaderRegistrationGenerator.cs

@@ -0,0 +1,83 @@
+using Goa.Clients.Dynamo.Generator.Models;
+
+namespace Goa.Clients.Dynamo.Generator.CodeGeneration;
+
+/// <summary>
+/// Generates a static registration class for all concrete DynamoDB model types.
+/// Consumers call <c>DynamoReaderRegistration.Initialize()</c> once at startup
+/// to trigger the static constructor, which registers all types with DynamoItemReaderRegistry.
+/// </summary>
+public class ReaderRegistrationGenerator : ICodeGenerator
+{
+    public string GenerateCode(IEnumerable<DynamoTypeInfo> types, GenerationContext context)
+    {
+        var registrableTypes = types.Where(ShouldRegister).ToList();
+
+        if (!registrableTypes.Any())
+            return string.Empty;
+
+        var typesByNamespace = registrableTypes.GroupBy(t => t.Namespace).Where(x => x.Any()).ToList();
+
+        if (!typesByNamespace.Any())
+            return string.Empty;
+
+        // Use the first namespace as the containing namespace for the registration class
+        var targetNamespace = typesByNamespace.First().Key;
+        if (string.IsNullOrEmpty(targetNamespace))
+            targetNamespace = "Generated";
+
+        var builder = new CodeBuilder();
+        builder.AppendLine("// <auto-generated/>");
+        builder.AppendLine("#nullable enable");
+        builder.AppendLine("using Goa.Clients.Dynamo;");
+        builder.AppendLine();
+        builder.AppendLine($"namespace {targetNamespace}");
+        builder.OpenBrace();
+        {
+            builder.OpenBraceWithLine("internal static class DynamoReaderRegistration");
+            {
+                builder.OpenBraceWithLine("static DynamoReaderRegistration()");
+                {
+                    foreach (var type in registrableTypes)
+                    {
+                        var normalizedName = NamingHelpers.NormalizeTypeName(type.Name);
+                        builder.AppendLine($"DynamoItemReaderRegistry.Register<{type.FullName}>(DynamoJsonMapper.{normalizedName}.ReadFromJson);");
+                        builder.AppendLine($"DynamoItemWriterRegistry.Register<{type.FullName}>(DynamoJsonMapper.{normalizedName}.WriteToJson);");
+                    }
+                }
+                builder.CloseBrace();
+                builder.AppendLine();
+                builder.AppendLine("/// <summary>");
+                builder.AppendLine("/// Ensures all DynamoDB model readers are registered. Call once during application startup.");
+                builder.AppendLine("/// </summary>");
+                builder.AppendLine("public static void Initialize() { }");
+            }
+            builder.CloseBrace();
+        }
+        builder.CloseBrace();
+
+        return builder.ToString();
+    }
+
+    private bool ShouldRegister(DynamoTypeInfo type)
+    {
+        if (type.IsAbstract)
+            return false;
+
+        return type.Attributes.Any(a => a is DynamoModelAttributeInfo) ||
+               HasInheritedDynamoModel(type);
+    }
+
+    private bool HasInheritedDynamoModel(DynamoTypeInfo type)
+    {
+        var current = type.BaseType;
+        while (current != null)
+        {
+            if (current.Attributes.Any(a => a is DynamoModelAttributeInfo))
+                return true;
+
+            current = current.BaseType;
+        }
+        return false;
+    }
+}

src/Clients/Goa.Clients.Dynamo.Generator/DynamoMapperIncrementalGenerator.cs

@@ -111,6 +111,8 @@ private static void Execute(Compilation compilation, ImmutableArray<INamedTypeSy
             var keyFactoryGenerator = new KeyFactoryGenerator(typeHandlerRegistry);
             var mapperGenerator = new MapperGenerator(typeHandlerRegistry);
             var extensionGenerator = new ExtensionGenerator(autoGenerateExtensions);
+            var jsonMapperGenerator = new JsonMapperGenerator(typeHandlerRegistry);
+            var readerRegistrationGenerator = new ReaderRegistrationGenerator();
 
             var keyFactoryCode = keyFactoryGenerator.GenerateCode(analyzedTypes, generationContext);
             var mapperCode = mapperGenerator.GenerateCode(analyzedTypes, generationContext);
@@ -124,6 +126,15 @@ private static void Execute(Compilation compilation, ImmutableArray<INamedTypeSy
             {
                 context.AddSource("DynamoExtensions.g.cs", SourceText.From(extensionCode, Encoding.UTF8));
             }
+
+            var jsonMapperCode = jsonMapperGenerator.GenerateCode(analyzedTypes, generationContext);
+            var readerRegistrationCode = readerRegistrationGenerator.GenerateCode(analyzedTypes, generationContext);
+
+            if (!string.IsNullOrEmpty(jsonMapperCode))
+                context.AddSource("DynamoJsonMapper.g.cs", SourceText.From(jsonMapperCode, Encoding.UTF8));
+
+            if (!string.IsNullOrEmpty(readerRegistrationCode))
+                context.AddSource("DynamoReaderRegistration.g.cs", SourceText.From(readerRegistrationCode, Encoding.UTF8));
         }
         catch (Exception ex)
         {

src/Clients/Goa.Clients.Dynamo.Generator/Models/DynamoConverterAttributeInfo.cs

@@ -0,0 +1,9 @@
+namespace Goa.Clients.Dynamo.Generator.Models;
+
+/// <summary>
+/// Represents DynamoConverter attribute information.
+/// </summary>
+public class DynamoConverterAttributeInfo : AttributeInfo
+{
+    public string ConverterTypeName { get; set; } = string.Empty;
+}

src/Clients/Goa.Clients.Dynamo.Generator/Models/PropertyInfo.cs

@@ -50,6 +50,11 @@ public ITypeSymbol UnderlyingType
     /// For non-dictionary types, returns null.
     /// </summary>
     public (ITypeSymbol KeyType, ITypeSymbol ValueType)? DictionaryTypes { get; set; }
+
+    /// <summary>
+    /// The fully qualified type name of a custom DynamoConverter for this property, if any.
+    /// </summary>
+    public string? ConverterTypeName { get; set; }
 
     /// <summary>
     /// Gets the effective name for this property in DynamoDB records.

src/Clients/Goa.Clients.Dynamo/DynamoExtensions.cs

@@ -218,6 +218,70 @@ public static async IAsyncEnumerable<DynamoRecord> ScanAllAsync(this IDynamoClie
         while (true);
     }
 
+    /// <summary>
+    /// Executes a typed DynamoDB Query operation with automatic pagination using an async iterator.
+    /// </summary>
+    public static async IAsyncEnumerable<T> QueryAllAsync<T>(this IDynamoClient client, string tableName, DynamoItemReader<T> itemReader, Action<QueryBuilder> builder, [EnumeratorCancellation] CancellationToken cancellationToken = default)
+    {
+        var _builder = new QueryBuilder(tableName);
+        builder(_builder);
+        var request = _builder.Build();
+
+        do
+        {
+            var result = await client.QueryAsync(request, itemReader, cancellationToken);
+            if (result.IsError)
+            {
+                throw new DynamoPaginationException(result.FirstError);
+            }
+
+            foreach (var item in result.Value.Items)
+            {
+                yield return item;
+            }
+
+            if (!result.Value.HasMoreResults)
+            {
+                break;
+            }
+
+            request.ExclusiveStartKey = result.Value.LastEvaluatedKey;
+        }
+        while (true);
+    }
+
+    /// <summary>
+    /// Executes a typed DynamoDB Scan operation with automatic pagination using an async iterator.
+    /// </summary>
+    public static async IAsyncEnumerable<T> ScanAllAsync<T>(this IDynamoClient client, string tableName, DynamoItemReader<T> itemReader, Action<ScanBuilder> builder, [EnumeratorCancellation] CancellationToken cancellationToken = default)
+    {
+        var _builder = new ScanBuilder(tableName);
+        builder(_builder);
+        var request = _builder.Build();
+
+        do
+        {
+            var result = await client.ScanAsync(request, itemReader, cancellationToken);
+            if (result.IsError)
+            {
+                throw new DynamoPaginationException(result.FirstError);
+            }
+
+            foreach (var item in result.Value.Items)
+            {
+                yield return item;
+            }
+
+            if (!result.Value.HasMoreResults)
+            {
+                break;
+            }
+
+            request.ExclusiveStartKey = result.Value.LastEvaluatedKey;
+        }
+        while (true);
+    }
+
     /// <summary>
     /// Executes a DynamoDB BatchGetItem operation with automatic retry of unprocessed keys using an async iterator.
     /// </summary>
@@ -227,9 +291,11 @@ public static async IAsyncEnumerable<DynamoRecord> ScanAllAsync(this IDynamoClie
     /// <returns>An async enumerable that yields items from all tables in the batch get result, retrying unprocessed keys.</returns>
     public static async IAsyncEnumerable<KeyValuePair<string, DynamoRecord>> BatchGetAllAsync(this IDynamoClient client, Action<BatchGetItemBuilder> builder, [EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
+        const int maxAttempts = 10;
         var _builder = new BatchGetItemBuilder();
         builder(_builder);
         var request = _builder.Build();
+        int attempts = 0;
 
         do
         {
@@ -252,8 +318,66 @@ public static async IAsyncEnumerable<KeyValuePair<string, DynamoRecord>> BatchGe
                 break;
             }
 
-            // Retry with unprocessed keys
+            if (attempts >= maxAttempts)
+            {
+                throw new DynamoPaginationException(
+                    ErrorOr.Error.Failure("Goa.DynamoDb.MaxRetriesExceeded", $"BatchGetItem still has unprocessed keys after {maxAttempts} retry attempts."));
+            }
+
+            // Retry with unprocessed keys using exponential backoff with jitter
+            request.RequestItems = result.Value.UnprocessedKeys!;
+            var baseDelay = Math.Min(100 * (1 << attempts), 25_600);
+            var jitter = Random.Shared.Next(0, baseDelay);
+            await Task.Delay(baseDelay + jitter, cancellationToken);
+            attempts++;
+        }
+        while (true);
+    }
+
+    /// <summary>
+    /// Executes a typed DynamoDB BatchGetItem operation with automatic retry of unprocessed keys using an async iterator.
+    /// </summary>
+    public static async IAsyncEnumerable<KeyValuePair<string, T>> BatchGetAllAsync<T>(this IDynamoClient client, DynamoItemReader<T> itemReader, Action<BatchGetItemBuilder> builder, [EnumeratorCancellation] CancellationToken cancellationToken = default)
+    {
+        const int maxAttempts = 10;
+        var _builder = new BatchGetItemBuilder();
+        builder(_builder);
+        var request = _builder.Build();
+        int attempts = 0;
+
+        do
+        {
+            var result = await client.BatchGetItemAsync(request, itemReader, cancellationToken);
+            if (result.IsError)
+            {
+                throw new DynamoPaginationException(result.FirstError);
+            }
+
+            foreach (var tableResponse in result.Value.Responses)
+            {
+                foreach (var item in tableResponse.Value)
+                {
+                    yield return new KeyValuePair<string, T>(tableResponse.Key, item);
+                }
+            }
+
+            if (!result.Value.HasUnprocessedKeys)
+            {
+                break;
+            }
+
+            if (attempts >= maxAttempts)
+            {
+                throw new DynamoPaginationException(
+                    ErrorOr.Error.Failure("Goa.DynamoDb.MaxRetriesExceeded", $"BatchGetItem still has unprocessed keys after {maxAttempts} retry attempts."));
+            }
+
+            // Retry with unprocessed keys using exponential backoff with jitter
             request.RequestItems = result.Value.UnprocessedKeys!;
+            var baseDelay = Math.Min(100 * (1 << attempts), 25_600);
+            var jitter = Random.Shared.Next(0, baseDelay);
+            await Task.Delay(baseDelay + jitter, cancellationToken);
+            attempts++;
         }
         while (true);
     }

src/Clients/Goa.Clients.Dynamo/DynamoItemReaderRegistry.cs

@@ -0,0 +1,28 @@
+namespace Goa.Clients.Dynamo;
+
+/// <summary>
+/// Registry for DynamoItemReader delegates, populated by source generators.
+/// </summary>
+public static class DynamoItemReaderRegistry
+{
+    /// <summary>
+    /// Registers a reader delegate for the specified type.
+    /// </summary>
+    /// <typeparam name="T">The type the reader deserializes.</typeparam>
+    /// <param name="reader">The reader delegate.</param>
+    public static void Register<T>(DynamoItemReader<T> reader) => Cache<T>.Reader = reader;
+
+    /// <summary>
+    /// Gets the registered reader delegate for the specified type.
+    /// </summary>
+    /// <typeparam name="T">The type to get the reader for.</typeparam>
+    /// <returns>The registered reader delegate.</returns>
+    /// <exception cref="InvalidOperationException">Thrown when no reader is registered for the type.</exception>
+    public static DynamoItemReader<T> Get<T>() => Cache<T>.Reader
+        ?? throw new InvalidOperationException($"No DynamoItemReader registered for {typeof(T).Name}. Ensure the type has [DynamoModel] attribute and the source generator has run.");
+
+    private static class Cache<T>
+    {
+        public static volatile DynamoItemReader<T>? Reader;
+    }
+}

src/Clients/Goa.Clients.Dynamo/DynamoItemWriterRegistry.cs

@@ -0,0 +1,28 @@
+namespace Goa.Clients.Dynamo;
+
+/// <summary>
+/// Registry for DynamoItemWriter delegates, populated by source generators.
+/// </summary>
+public static class DynamoItemWriterRegistry
+{
+    /// <summary>
+    /// Registers a writer delegate for the specified type.
+    /// </summary>
+    /// <typeparam name="T">The type the writer serializes.</typeparam>
+    /// <param name="writer">The writer delegate.</param>
+    public static void Register<T>(DynamoItemWriter<T> writer) => Cache<T>.Writer = writer;
+
+    /// <summary>
+    /// Gets the registered writer delegate for the specified type.
+    /// </summary>
+    /// <typeparam name="T">The type to get the writer for.</typeparam>
+    /// <returns>The registered writer delegate.</returns>
+    /// <exception cref="InvalidOperationException">Thrown when no writer is registered for the type.</exception>
+    public static DynamoItemWriter<T> Get<T>() => Cache<T>.Writer
+        ?? throw new InvalidOperationException($"No DynamoItemWriter registered for {typeof(T).Name}. Ensure the type has [DynamoModel] attribute and the source generator has run.");
+
+    private static class Cache<T>
+    {
+        public static volatile DynamoItemWriter<T>? Writer;
+    }
+}