shoc-backend/Api.SeaHavenIndustries/Infrastructure/DynamoSyncExternalSource.cs

using Amazon.DynamoDBv2;
using Amazon.DynamoDBv2.Model;
using SeaHaven.Services.Interfaces;

namespace Api.SeaHavenIndustries.Infrastructure
{
    public class DynamoSyncExternalSource : ISyncExternalSource
    {
        private readonly AmazonDynamoDBClient _dynamo;
        private const string WO_TABLE = "WorkOrders";
        private const string COMMENTS_TABLE = "WorkOrderComments";
        private const string VENDOR_REPLIES_TABLE = "VendorReplies";

        public DynamoSyncExternalSource()
        {
            _dynamo = new AmazonDynamoDBClient(Amazon.RegionEndpoint.USEast1);
        }

        public async IAsyncEnumerable<Dictionary<string, string?>> ScanWorkOrdersAsync(
            [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken)
        {
            Dictionary<string, AttributeValue>? lastKey = null;
            do
            {
                cancellationToken.ThrowIfCancellationRequested();
                var request = new ScanRequest
                {
                    TableName = WO_TABLE,
                    Limit = 100,
                    ExclusiveStartKey = lastKey
                };
                var response = await _dynamo.ScanAsync(request);
                foreach (var item in response.Items)
                    yield return ConvertItem(item);
                lastKey = response.LastEvaluatedKey;
            } while (lastKey != null && lastKey.Count > 0);
        }

        public async IAsyncEnumerable<Dictionary<string, string?>> ScanCommentsAsync(
            [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken)
        {
            Dictionary<string, AttributeValue>? lastKey = null;
            do
            {
                cancellationToken.ThrowIfCancellationRequested();
                var request = new ScanRequest
                {
                    TableName = COMMENTS_TABLE,
                    Limit = 100,
                    ExclusiveStartKey = lastKey
                };
                var response = await _dynamo.ScanAsync(request);
                foreach (var item in response.Items)
                    yield return ConvertItem(item);
                lastKey = response.LastEvaluatedKey;
            } while (lastKey != null && lastKey.Count > 0);
        }

        public async IAsyncEnumerable<Dictionary<string, string?>> ScanVendorRepliesAsync(
            [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken)
        {
            Dictionary<string, AttributeValue>? lastKey = null;
            do
            {
                cancellationToken.ThrowIfCancellationRequested();
                var request = new ScanRequest
                {
                    TableName = VENDOR_REPLIES_TABLE,
                    Limit = 100,
                    ExclusiveStartKey = lastKey
                };
                var response = await _dynamo.ScanAsync(request);
                foreach (var item in response.Items)
                    yield return ConvertItem(item);
                lastKey = response.LastEvaluatedKey;
            } while (lastKey != null && lastKey.Count > 0);
        }

        private static Dictionary<string, string?> ConvertItem(Dictionary<string, AttributeValue> item)
        {
            var result = new Dictionary<string, string?>();
            foreach (var kvp in item)
            {
                if (kvp.Value.S != null)
                    result[kvp.Key] = kvp.Value.S;
            }
            return result;
        }
    }
}