Revamp and simplify conversation management

readme.md

@@ -16,7 +16,7 @@ Create agents for WhatsApp using Azure Functions.
 var builder = FunctionsApplication.CreateBuilder(args);
 builder.ConfigureFunctionsWebApplication();
 
-builder.Services.AddWhatsApp<MyWhatsAppHandler>(builder.Configuration);
+builder.Services.AddWhatsApp<MyWhatsAppHandler>();
 
 builder.Build().Run();
 ```
@@ -25,7 +25,7 @@ Instead of providing an `IWhatsAppHandler` implementation, you can also
 register an inline handler using minimal API style:
 
 ```csharp
-builder.Services.AddWhatsApp(builder.Configuration, (messages, cancellation) =>
+builder.Services.AddWhatsApp((messages, cancellation) =>
 {
     foreach (var message in messages)
     {
@@ -56,7 +56,7 @@ If the handler needs additional services, they can be provided directly
 as generic parameters of the `UseWhatsApp` method, such as:
 
 ```csharp
-builder.Services.AddWhatsApp<ILogger<Program>>(builder.Configuration, (logger, message, cancellation) =>
+builder.Services.AddWhatsApp<ILogger<Program>>((logger, message, cancellation) =>
 {
     logger.LogInformation($"Got messages!");
 
@@ -121,6 +121,44 @@ The above code would render as follows in WhatsApp:
 
 ![](https://raw.githubusercontent.com/devlooped/WhatsApp/main/assets/img/whatsapp.png)
 
+## Conversations
+
+WhatsApp does not provide a way to keep track of conversations, at most providing the 
+related message ID of a message that was replied to. In many agents, however, keeping 
+track of conversations is crucial for maintaining context and continuity. 
+
+This library provides a simple built-in functionality for this based on some simple 
+heuristics: 
+
+- If a message is sent in response to another message, it is considered part of the same conversation.
+- Messages sent within a short time frame (default: 5 minutes) are considered part of the same conversation.
+- Individual messages, conversations and the active conversations are stored in an Azure 
+  storage account
+
+Usage:
+
+```csharp
+builder.Services
+    .AddWhatsApp<MyWhatsAppHandler>()
+    .UseConversation(conversationWindowSeconds: 300 /* default */);
+```
+
+Unless you provide a [CloudStorageAccount](https://www.nuget.org/packages/Devlooped.CloudStorageAccount) in 
+the service collection, the library will use the `AzureWebJobsStorage` connection string automatically 
+for this, so things will just work out of the box. 
+
+An example of providing storage to a different account than the functions runtime one:
+
+```csharp
+builder.Services.AddSingleton(services => builder.Environment.IsDevelopment() ?
+    // Always local emulator in development
+    CloudStorageAccount.DevelopmentStorageAccount :
+    // First try with custom connection string
+    CloudStorageAccount.TryParse(builder.Configuration["App:Storage"] ?? "", out var storage) ?
+    storage :
+    // Fallback to built-in functions storage (default behavior).
+    CloudStorageAccount.Parse(builder.Configuration["AzureWebJobsStorage"]));
+```
 
 ## Configuration
 
@@ -176,7 +214,7 @@ message storage and conversation management:
 var builder = FunctionsApplication.CreateBuilder(args);
 builder.ConfigureFunctionsWebApplication();
 
-builder.Services.AddWhatsApp<MyWhatsAppHandler>(builder.Configuration)
+builder.Services.AddWhatsApp<MyWhatsAppHandler>()
     .UseOpenTelemetry(builder.Environment.ApplicationName)
     .UseLogging()
     .UseStorage()
@@ -220,7 +258,7 @@ This new extension method can now be used in the pipeline without changing any o
 existing handlers:
 
 ```csharp
-builder.Services.AddWhatsApp<MyWhatsAppHandler>(builder.Configuration)
+builder.Services.AddWhatsApp<MyWhatsAppHandler>()
     .UseOpenTelemetry(builder.Environment.ApplicationName)
     .UseLogging()
     .UseIgnore()  // 👈 Ignore status+unsupported messages. We do log them.

src/Console/Program.cs

@@ -24,8 +24,15 @@
 host.Configuration.AddDotNetConfig();
 host.Configuration.AddUserSecrets<Program>();
 
-host.Services.AddHttpClient("whatsapp")
-    .AddStandardResilienceHandler();
+var http = host.Services.AddHttpClient("whatsapp");
+if (Debugger.IsAttached)
+{
+    http.ConfigureHttpClient(http => http.Timeout = TimeSpan.FromHours(1));
+}
+else
+{
+    http.AddStandardResilienceHandler();
+}
 
 var cts = new CancellationTokenSource();
 Console.CancelKeyPress += (s, e) => cts.Cancel();

src/SampleApp/Sample/Program.cs

@@ -38,14 +38,13 @@
     CloudStorageAccount.DevelopmentStorageAccount :
     CloudStorageAccount.TryParse(builder.Configuration["App:Storage"] ?? "", out var storage) ?
     storage :
-    throw new InvalidOperationException("Missing required App:Storage connection string."));
+    CloudStorageAccount.Parse(builder.Configuration["AzureWebJobsStorage"]));
 
 builder.Services
-    .AddWhatsApp<ProcessHandler>(builder.Configuration)
+    .AddWhatsApp<ProcessHandler>()
     // Matches what we use in ConfigureOpenTelemetry
     .UseOpenTelemetry(builder.Environment.ApplicationName)
     .UseLogging()
-    .UseStorage()
-    .UseConversation();
+    .UseConversation(conversationWindowSeconds: 300 /* default */);
 
 builder.Build().Run();

src/Tests/IntegrationTests.cs

@@ -39,12 +39,11 @@ public async Task RunConversationAsync()
         IEnumerable<IMessage>? messages = null;
 
         services
-            .AddWhatsApp(configuration, (input, cancellation) =>
+            .AddWhatsApp((input, cancellation) =>
             {
                 messages = input.ToArray();
                 return AsyncEnumerable.Empty<Response>();
             })
-            .UseStorage()
             .UseConversation();
 
         var provider = services.BuildServiceProvider();

src/Tests/PipelineTests.cs

@@ -1,4 +1,8 @@
-namespace Devlooped.WhatsApp;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+using Moq;
+
+namespace Devlooped.WhatsApp;
 
 public class PipelineTests(ITestOutputHelper output)
 {
@@ -70,4 +74,157 @@ public async Task CanBuildLoggingPipeline()
         Assert.True(after);
         Assert.True(target);
     }
+
+    [Fact]
+    public async Task ConversationCalledAfterCustom()
+    {
+        var configuration = new ConfigurationBuilder()
+            .AddInMemoryCollection(new Dictionary<string, string?>()
+            {
+                { "Meta:VerifyToken", "test-challenge" },
+                { "Meta:Numbers:1234567890", "test-access-token" }
+            })
+            .Build();
+
+        IAsyncEnumerable<Response> messages = new Response[]
+        {
+            new TextResponse("123", "456", "Foo", "Bar", "Baz")
+        }.ToAsyncEnumerable();
+
+        var order = new List<string>();
+
+        var handler = Mock.Of<IWhatsAppHandler>(x => x.HandleAsync(It.IsAny<IEnumerable<IMessage>>(), It.IsAny<CancellationToken>()) == messages);
+        var conversation = new Mock<IConversationStorage>();
+
+        conversation
+            .Setup(x => x.GetMessageAsync(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<CancellationToken>()))
+            .Callback(() => order.Add("storage:read"));
+
+        conversation
+            .Setup(x => x.GetMessagesAsync(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<CancellationToken>()))
+            .Returns<string, string, CancellationToken>((number, conversation, cancellation) =>
+            {
+                order.Add("storage:all");
+                return messages;
+            });
+
+        conversation
+            .Setup(x => x.GetActiveConversationAsync(It.IsAny<string>(), It.IsAny<CancellationToken>()))
+            .Callback(() => order.Add("storage:active"));
+
+        conversation
+            .Setup(x => x.SaveAsync(It.IsAny<IMessage>(), It.IsAny<CancellationToken>()))
+            .Callback(() => order.Add("storage:save"));
+
+        var services = new ServiceCollection()
+            .AddSingleton<IConfiguration>(configuration)
+            .AddSingleton(conversation.Object);
+
+        services.AddWhatsApp(handler)
+            .Use((messages, inner, cancellation) =>
+            {
+                order.Add("before");
+                return inner.HandleAsync(messages, cancellation);
+            })
+            .UseConversation()
+            .Use((messages, inner, cancellation) =>
+            {
+                order.Add("after");
+                return inner.HandleAsync(messages, cancellation);
+            });
+
+        // Override default IWhatsAppClient to prevent any actual sending
+        var client = new Mock<IWhatsAppClient>();
+        client.Setup(x => x.SendAsync(It.IsAny<string>(), It.IsAny<object>(), It.IsAny<CancellationToken>()))
+              .Callback(() => order.Add("client:send"));
+
+        services.AddSingleton<IWhatsAppClient>(client.Object);
+
+        var provider = services.BuildServiceProvider();
+        await provider.GetRequiredService<IWhatsAppHandler>()
+            .HandleAsync(new ReactionMessage("msgid", service, user, 0, "🗽"));
+
+        //             custom    👇 get convo id   👇 input 💾   👇 convo 🔎    custom   👇 output 📨  👇 response 💾
+        Assert.Equal(["before", "storage:active", "storage:save", "storage:all", "after", "client:send", "storage:save"], order);
+    }
+
+    [Fact]
+    public async Task ConversationRestored()
+    {
+        var configuration = new ConfigurationBuilder()
+            .AddInMemoryCollection(new Dictionary<string, string?>()
+            {
+                { "Meta:VerifyToken", "test-challenge" },
+                { "Meta:Numbers:1234567890", "test-access-token" }
+            })
+            .Build();
+
+        var storage = new MemoryConversationStorage();
+        var response = AsyncEnum<Response>([new TextResponse(user.Number, service.Id, "1234", null, "Bye")]);
+        var messages = new List<IMessage[]>();
+
+        var handler = new Mock<IWhatsAppHandler>();
+        handler.Setup(x => x.HandleAsync(It.IsAny<IEnumerable<IMessage>>(), It.IsAny<CancellationToken>()))
+            .Returns((IEnumerable<IMessage> input, CancellationToken _) =>
+            {
+                messages.Add(input.ToArray());
+                // Timestamp in WhatsApp is in Unix seconds, so we need to simulate a delay
+                Thread.Sleep(1000);
+                var message = input.OfType<ContentMessage>().Last();
+                return AsyncEnum<Response>([message.Reply(message.Content.ToString() + " Reply")]);
+            });
+
+        var services = new ServiceCollection()
+            .AddSingleton<IConfiguration>(configuration)
+            .AddSingleton<IConversationStorage>(storage);
+
+        services.AddWhatsApp(handler.Object).UseConversation();
+
+        // Override default IWhatsAppClient to prevent any actual sending
+        var client = new Mock<IWhatsAppClient>();
+        client.Setup(x => x.SendAsync(It.IsAny<string>(), It.IsAny<object>(), It.IsAny<CancellationToken>()))
+              .ReturnsAsync(Ulid.NewUlid().ToString());
+
+        services.AddSingleton<IWhatsAppClient>(client.Object);
+
+        var provider = services.BuildServiceProvider();
+        var pipeline = provider.GetRequiredService<IWhatsAppHandler>();
+
+        await pipeline.HandleAsync(Text("Hello"));
+
+        Assert.Single(messages);
+        Assert.Single(messages[0]);
+
+        // Timestamps in WhatsApp are in Unix seconds, so we need to wait.
+        Thread.Sleep(1000);
+        await pipeline.HandleAsync(Text("Hello Again"));
+
+        Assert.Equal(2, messages.Count);
+        // Second messages now contain first message, its response and the new one
+        Assert.Equal(3, messages[1].Length);
+        Assert.IsAssignableFrom<ContentMessage>(messages[1][0]);
+        Assert.IsAssignableFrom<Response>(messages[1][1]);
+        Assert.IsAssignableFrom<ContentMessage>(messages[1][2]);
+    }
+
+    ContentMessage Text(string text) => new ContentMessage(
+        Ulid.NewUlid().ToString(), service, user, DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(), new TextContent(text));
+
+    IAsyncEnumerable<IMessage> AsyncEnum(IEnumerable<IMessage> messages) => messages.ToAsyncEnumerable();
+
+    IAsyncEnumerable<T> AsyncEnum<T>(IEnumerable<T> messages) => messages.ToAsyncEnumerable();
+
+    class MemoryConversationStorage : ConversationStorage
+    {
+        public MemoryConversationStorage() : base(CloudStorageAccount.DevelopmentStorageAccount) { }
+
+        protected override IDocumentRepository<Conversation> CreateActiveConversationRepository()
+            => MemoryRepository.Create<Conversation>(rowKey: _ => "active");
+
+        protected override IDocumentRepository<Conversation> CreateConversationsRepository()
+            => MemoryRepository.Create<Conversation>();
+
+        protected override IDocumentRepository<IMessage> CreateMessagesRepository()
+            => MemoryRepository.Create<IMessage>("messages", x => x.UserNumber, x => x.Id);
+    }
 }

src/Tests/Tests.csproj

@@ -11,6 +11,7 @@
 
   <ItemGroup>
     <PackageReference Include="coverlet.collector" Version="6.0.4" />
+    <PackageReference Include="Devlooped.TableStorage.Memory" Version="5.3.1" />
     <PackageReference Include="Microsoft.CodeAnalysis.CSharp.Analyzer.Testing" Version="1.1.2" />
     <PackageReference Include="Microsoft.CodeAnalysis.CSharp.Workspaces" Version="4.13.0" />
     <PackageReference Include="Microsoft.Extensions.Configuration" Version="8.0.0" />
@@ -21,6 +22,7 @@
     <PackageReference Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="8.0.0" />
     <PackageReference Include="Microsoft.Extensions.Options.DataAnnotations" Version="8.0.0" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.14.1" />
+    <PackageReference Include="Moq" Version="4.20.72" />
     <PackageReference Include="xunit" Version="2.9.3" />
     <PackageReference Include="xunit.runner.visualstudio" Version="3.1.1" />
   </ItemGroup>

src/Tests/WhatsAppHandlerExtensions.cs

@@ -2,6 +2,12 @@
 
 public static class WhatsAppHandlerExtensions
 {
+    /// <summary>
+    /// Test-only extension method to handle a single message and force enumeration of all 
+    /// responses for the purpose of full pipeline execution only. It also sets the timestamp 
+    /// of the message to the current UTC time.
+    /// </summary>
     public static Task HandleAsync(this IWhatsAppHandler handler, Message message, CancellationToken cancellation = default)
-        => handler.HandleAsync([message], cancellation).ForEachAsync(x => { }, cancellation);
+        => handler.HandleAsync([message with { Timestamp = DateTimeOffset.UtcNow.ToUnixTimeSeconds() }], cancellation)
+            .ForEachAsync(x => { }, cancellation);
 }

src/WhatsApp/Content.cs

@@ -35,6 +35,8 @@ public record ContactContent(string Name, string Surname, string[] Numbers) : Co
     /// <inheritdoc/>
     [JsonIgnore]
     public override ContentType Type => ContentType.Contact;
+
+    public override string ToString() => $"{Name} {Surname} ({string.Join(", ", Numbers)})";
 }
 
 /// <summary>
@@ -46,6 +48,8 @@ public record TextContent(string Text) : Content
     /// <inheritdoc/>
     [JsonIgnore]
     public override ContentType Type => ContentType.Text;
+
+    public override string ToString() => Text;
 }
 
 /// <summary>

src/WhatsApp/Conversation.cs

@@ -12,4 +12,5 @@
 /// <param name="Id">The unique identifier for the conversation. Cannot be null or empty.</param>
 /// <param name="Messages">A list of messages exchanged in the conversation. Cannot be null; may be empty if no messages exist.</param>
 /// <param name="Timestamp">The timestamp of the conversation, represented as the number of milliseconds since the Unix epoch.</param>
-public record Conversation(string Number, string Id, List<IMessage> Messages, long Timestamp);
+[Table("WhatsAppConversations")]
+public record Conversation([PartitionKey] string Number, [RowKey] string Id, List<IMessage> Messages, long Timestamp);