Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,10 @@ Versioning: [Semantic Versioning](https://semver.org/spec/v2.0.0.html).

## [Unreleased]

### Fixed

- **Message history can be read again on PostgreSQL.** Every read of a stored message failed with a server error (500). Postgres stores message content as `jsonb`, which reorders the keys of each block, and the reader required the block's type to come first. The bug was latent since message storage first shipped (1.2.0). It broke `GET /api/v1/conversations/{id}/messages`, `GET /api/v1/supervisor/conversations/{id}/messages`, typification suggestions, the GDPR export, the conversation summary report, delivery-status webhooks, de-duplication of redelivered inbound messages and email thread matching. The in-memory store was not affected. No migration is needed: rows already stored are read correctly after the upgrade.

### Dependencies

- **Verbara.Sdk `2.4.0` → `2.7.0`** and **Verbara.Sdk.Pro `2.14.1-pro` → `2.17.0-pro`** — all 29
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,10 @@ namespace Verbara.Platform.Conversations.Serialization;
[JsonSerializable(typeof(MessageDirection))]
[JsonSerializable(typeof(MessageDeliveryStatus))]
[JsonSerializable(typeof(MessageBlockType))]
// AllowOutOfOrderMetadataProperties: messages.content is JSONB, and Postgres re-orders object
// keys, so a MessageBlock's "$type" discriminator is not the first property when read back.
[JsonSourceGenerationOptions(
PropertyNamingPolicy = JsonKnownNamingPolicy.CamelCase,
DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull)]
DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull,
AllowOutOfOrderMetadataProperties = true)]
public partial class ConversationsJsonContext : JsonSerializerContext;
Original file line number Diff line number Diff line change
Expand Up @@ -83,18 +83,23 @@ namespace Verbara.Platform.Storage.Postgres;
// reflection) under JsonSerializerIsReflectionEnabledByDefault=false (design D2, D6).
[JsonSerializable(typeof(IReadOnlyList<QuiescedCampaignInfo>))]
[JsonSerializable(typeof(QuiescedCampaignInfo))]
// AllowOutOfOrderMetadataProperties: JSONB re-orders object keys, so a polymorphic "$type"
// discriminator (MessageBlock, via MessageEnvelope) is not the first property when read back.
[JsonSourceGenerationOptions(
PropertyNamingPolicy = JsonKnownNamingPolicy.CamelCase,
DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull,
UseStringEnumConverter = true)]
UseStringEnumConverter = true,
AllowOutOfOrderMetadataProperties = true)]
internal sealed partial class PostgresJsonContext : JsonSerializerContext;

internal static class PostgresJson
{
// These options replace the attribute's wholesale, so they repeat what JSONB needs.
internal static readonly PostgresJsonContext Ctx = new(new JsonSerializerOptions
{
PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull,
AllowOutOfOrderMetadataProperties = true,
});

internal static string Serialize<T>(T value, System.Text.Json.Serialization.Metadata.JsonTypeInfo<T> typeInfo) =>
Expand Down
20 changes: 20 additions & 0 deletions tests/Verbara.Platform.Conversations.Tests/MessageBlockTests.cs
Original file line number Diff line number Diff line change
Expand Up @@ -166,4 +166,24 @@ public void Conversation_ShouldExposeChannel_WhenCreated()

conv.Channel.Should().Be(ChannelType.WhatsApp);
}

[Fact]
public void MessageEnvelope_ShouldDeserialize_WhenTypeDiscriminatorIsNotFirst()
{
// Postgres JSONB re-orders object keys (shorter keys first), so a block read back from
// messages.content carries "$type" after its other properties.
const string json = """
{"blocks": [
{"text": "hello", "type": 0, "$type": "text"},
{"url": "https://x/i.jpg", "type": 1, "$type": "image", "caption": "c", "mimeType": "image/jpeg"}
]}
""";

var envelope = JsonSerializer.Deserialize(json, ConversationsJsonContext.Default.MessageEnvelope);

envelope.Should().NotBeNull();
envelope!.Blocks.Should().HaveCount(2);
envelope.Blocks[0].Should().BeOfType<TextBlock>().Which.Text.Should().Be("hello");
envelope.Blocks[1].Should().BeOfType<ImageBlock>().Which.Url.Should().Be("https://x/i.jpg");
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,33 @@
using Verbara.Platform.Conversations;

namespace Verbara.Platform.Storage.Postgres.Tests;

/// <summary>
/// <see cref="PostgresJson.Ctx"/> is the shared context for JSONB columns, and JSONB re-orders
/// object keys. <see cref="MessageEnvelope"/> is registered on it and carries the polymorphic
/// <see cref="MessageBlock"/>, whose <c>$type</c> discriminator is not first once read back.
/// </summary>
public class PostgresJsonContextTests
{
[Fact]
public void Deserialize_ShouldReadPolymorphicBlocks_WhenTypeDiscriminatorIsNotFirst()
{
const string jsonbText = """
{"blocks": [{"text": "hello", "type": 0, "$type": "text"},
{"url": "https://x/i.jpg", "type": 1, "$type": "image", "mimeType": "image/jpeg"}]}
""";

var envelope = PostgresJson.Deserialize(jsonbText, PostgresJson.Ctx.MessageEnvelope);

envelope.Should().NotBeNull();
envelope!.Blocks.Select(b => b.GetType()).Should().Equal(typeof(TextBlock), typeof(ImageBlock));
}

[Fact]
public void Ctx_ShouldAllowOutOfOrderMetadata_WhenConstructedWithExplicitOptions()
{
// The explicit options passed to the PostgresJsonContext constructor replace the
// [JsonSourceGenerationOptions] attribute wholesale, so both must carry the setting.
PostgresJson.Ctx.Options.AllowOutOfOrderMetadataProperties.Should().BeTrue();
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,84 @@
using DotNet.Testcontainers.Builders;
using DotNet.Testcontainers.Containers;
using Npgsql;

namespace Verbara.Platform.Storage.Postgres.Tests.Stores;

/// <summary>
/// Testcontainers-backed Postgres fixture for <see cref="PostgresMessageStoreJsonbTests"/>.
/// Spins up <c>postgres:18-alpine</c> (the production major) with the <c>messages</c> DDL
/// mirrored from migration <c>001_Baseline.sql</c>, so the <c>content</c> JSONB column
/// re-orders object keys exactly as it does in production.
/// </summary>
public sealed class MessageStoreFixture : IAsyncLifetime
{
private IContainer? _container;

public string ConnectionString =>
$"Host={_container!.Hostname};Port={_container.GetMappedPublicPort(5432)};" +
"Database=postgres;Username=postgres;Password=postgres";

public NpgsqlDataSource DataSource { get; private set; } = null!;

public async Task InitializeAsync()
{
_container = new ContainerBuilder("postgres:18-alpine")
.WithEnvironment("POSTGRES_PASSWORD", "postgres")
.WithEnvironment("POSTGRES_DB", "postgres")
.WithPortBinding(5432, true)
.WithWaitStrategy(
Wait.ForUnixContainer()
// `-h 127.0.0.1` forces the readiness probe over TCP — see
// ConversationVoiceLinkFixture for why a socket-only probe is unsafe.
.UntilCommandIsCompleted("pg_isready", "-U", "postgres", "-h", "127.0.0.1"))
.Build();

await _container.StartAsync();

DataSource = NpgsqlDataSource.Create(ConnectionString);

await using var conn = await DataSource.OpenConnectionAsync();
await using var cmd = conn.CreateCommand();
cmd.CommandText = SchemaSql;
await cmd.ExecuteNonQueryAsync();
}

public async Task DisposeAsync()
{
if (DataSource is not null) await DataSource.DisposeAsync();
if (_container is not null) await _container.DisposeAsync();
}

/// <summary>Truncate the table between tests for isolation.</summary>
public async Task ResetAsync()
{
await using var conn = await DataSource.OpenConnectionAsync();
await using var cmd = conn.CreateCommand();
cmd.CommandText = "TRUNCATE messages";
await cmd.ExecuteNonQueryAsync();
}

// messages DDL — verbatim subset of the consolidated 001_Baseline.sql.
private const string SchemaSql = """
CREATE TABLE IF NOT EXISTS messages (
message_id TEXT NOT NULL,
conversation_id TEXT NOT NULL,
tenant_id TEXT NOT NULL,
direction INTEGER NOT NULL,
channel INTEGER NOT NULL,
sender_id TEXT,
content JSONB NOT NULL,
delivery_status INTEGER NOT NULL DEFAULT 0,
external_message_id TEXT,
created_at TIMESTAMPTZ NOT NULL,
delivered_at TIMESTAMPTZ,
read_at TIMESTAMPTZ,
updated_at TIMESTAMPTZ,
created_by TEXT,
updated_by TEXT,
PRIMARY KEY (tenant_id, message_id)
);
CREATE INDEX IF NOT EXISTS idx_messages_conversation ON messages (tenant_id, conversation_id, created_at);
CREATE INDEX IF NOT EXISTS idx_messages_external ON messages (tenant_id, external_message_id);
""";
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,184 @@
using Npgsql;
using Verbara.Platform.Conversations;
using Verbara.Platform.Core;
using Verbara.Platform.Storage.Postgres.Stores;

namespace Verbara.Platform.Storage.Postgres.Tests.Stores;

/// <summary>
/// Reads message history back from a real Postgres <c>messages.content</c> JSONB column.
/// JSONB does not keep object key order, so the <see cref="MessageBlock"/> type
/// discriminator <c>$type</c> is no longer the first property of each block when it is read
/// back. Before 2.25.1 every read below threw <see cref="NotSupportedException"/>
/// ("must specify a type discriminator") from the source-generated deserializer.
/// </summary>
public class PostgresMessageStoreJsonbTests : IClassFixture<MessageStoreFixture>, IAsyncLifetime
{
private readonly MessageStoreFixture _fixture;
private readonly PostgresMessageStore _sut;
private readonly TenantId _tenant;

public PostgresMessageStoreJsonbTests(MessageStoreFixture fixture)
{
_fixture = fixture;
_sut = new PostgresMessageStore(_fixture.DataSource);
_tenant = new TenantId($"t-{Guid.NewGuid():N}");
}

public async Task InitializeAsync() => await _fixture.ResetAsync();

public Task DisposeAsync() => Task.CompletedTask;

private Message MakeMessage(
EntityId conversationId, string? externalId = null, DateTimeOffset? createdAt = null, MessageEnvelope? content = null) =>
new()
{
MessageId = EntityId.New(),
ConversationId = conversationId,
TenantId = _tenant,
Direction = MessageDirection.Inbound,
Channel = ChannelType.WhatsApp,
SenderId = "contact-1",
Content = content ?? new MessageEnvelope(
[
new TextBlock("hello"),
new ImageBlock("https://cdn.example.com/a.jpg", "a photo", "image/jpeg"),
]),
DeliveryStatus = MessageDeliveryStatus.Delivered,
ExternalMessageId = externalId,
CreatedAt = createdAt ?? DateTimeOffset.UtcNow,
};

private static void AssertTextAndImage(Message message)
{
message.Content.Blocks.Should().HaveCount(2);
message.Content.Blocks[0].Should().BeOfType<TextBlock>().Which.Text.Should().Be("hello");
var image = message.Content.Blocks[1].Should().BeOfType<ImageBlock>().Subject;
image.Url.Should().Be("https://cdn.example.com/a.jpg");
image.Caption.Should().Be("a photo");
image.MimeType.Should().Be("image/jpeg");
}

[Fact]
public async Task SaveAsync_ShouldStoreDiscriminatorOutOfFirstPosition_WhenContentIsJsonb()
{
// Pins the precondition the read tests depend on: if Postgres ever kept key order,
// the tests below would pass without exercising the bug.
var message = MakeMessage(EntityId.New());
await _sut.SaveAsync(message, CancellationToken.None);

await using var conn = await _fixture.DataSource.OpenConnectionAsync();
await using var cmd = conn.CreateCommand();
cmd.CommandText = "SELECT (content->'blocks'->0)::text FROM messages WHERE tenant_id = @t AND message_id = @m";
cmd.Parameters.Add(new NpgsqlParameter("t", _tenant.Value));
cmd.Parameters.Add(new NpgsqlParameter("m", message.MessageId.Value));

var storedBlock = (string?)await cmd.ExecuteScalarAsync();

storedBlock.Should().Contain("\"$type\": \"text\"");
storedBlock.Should().NotStartWith("{\"$type\"");
}

[Fact]
public async Task GetConversationMessagesAsync_ShouldReturnTextAndImageBlocks_WhenContentIsJsonb()
{
var conversationId = EntityId.New();
await _sut.SaveAsync(MakeMessage(conversationId), CancellationToken.None);

var messages = await _sut.GetConversationMessagesAsync(_tenant, conversationId, 50, 0, CancellationToken.None);

AssertTextAndImage(messages.Should().ContainSingle().Subject);
}

[Fact]
public async Task GetByIdAsync_ShouldReturnTextAndImageBlocks_WhenContentIsJsonb()
{
var message = MakeMessage(EntityId.New());
await _sut.SaveAsync(message, CancellationToken.None);

var read = await _sut.GetByIdAsync(_tenant, message.MessageId, CancellationToken.None);

read.Should().NotBeNull();
AssertTextAndImage(read!);
}

[Fact]
public async Task FindByExternalIdAsync_ShouldReturnTextAndImageBlocks_WhenContentIsJsonb()
{
var message = MakeMessage(EntityId.New(), externalId: "wamid.HBgM-roundtrip");
await _sut.SaveAsync(message, CancellationToken.None);

var read = await _sut.FindByExternalIdAsync(_tenant, "wamid.HBgM-roundtrip", CancellationToken.None);

read.Should().NotBeNull();
read!.MessageId.Should().Be(message.MessageId);
AssertTextAndImage(read);
}

[Fact]
public async Task GetByConversationIdsAsync_ShouldReturnTextAndImageBlocks_WhenContentIsJsonb()
{
var first = EntityId.New();
var second = EntityId.New();
var now = DateTimeOffset.UtcNow;
await _sut.SaveAsync(MakeMessage(first, createdAt: now), CancellationToken.None);
await _sut.SaveAsync(MakeMessage(second, createdAt: now.AddSeconds(1)), CancellationToken.None);

var messages = await _sut.GetByConversationIdsAsync(_tenant, [first, second], CancellationToken.None);

messages.Should().HaveCount(2);
messages.Select(m => m.ConversationId).Should().Equal(first, second);
messages.Should().AllSatisfy(AssertTextAndImage);
}

[Fact]
public async Task GetByIdAsync_ShouldReturnEveryBlockType_WhenContentIsJsonb()
{
var message = MakeMessage(EntityId.New(), content: new MessageEnvelope(
[
new TextBlock("t"),
new ImageBlock("https://x/i.png", null, "image/png"),
new AudioBlock("https://x/a.ogg", TimeSpan.FromSeconds(7), "audio/ogg"),
new VideoBlock("https://x/v.mp4", "clip", null),
new FileBlock("https://x/f.pdf", "f.pdf", "application/pdf", 1024L),
new LocationBlock(4.65, -74.05, "Bogota"),
new InteractiveBlock("pick one", [new QuickReply("y", "Yes"), new QuickReply("n", "No")]),
]));
await _sut.SaveAsync(message, CancellationToken.None);

var read = await _sut.GetByIdAsync(_tenant, message.MessageId, CancellationToken.None);

read.Should().NotBeNull();
read!.Content.Blocks.Select(b => b.GetType()).Should().Equal(
typeof(TextBlock), typeof(ImageBlock), typeof(AudioBlock), typeof(VideoBlock),
typeof(FileBlock), typeof(LocationBlock), typeof(InteractiveBlock));
read.Content.Blocks[2].Should().BeOfType<AudioBlock>().Which.Duration.Should().Be(TimeSpan.FromSeconds(7));
read.Content.Blocks[5].Should().BeOfType<LocationBlock>().Which.Name.Should().Be("Bogota");
read.Content.Blocks[6].Should().BeOfType<InteractiveBlock>().Which.Replies.Should().HaveCount(2);
}

[Fact]
public async Task GetByIdAsync_ShouldReadRow_WhenWrittenByEarlierRelease()
{
// Rows persisted by <= 2.25.0 are stored exactly like this (JSONB key order); no
// migration is needed — the fix is on the read side only.
var messageId = EntityId.New();
await using (var conn = await _fixture.DataSource.OpenConnectionAsync())
await using (var cmd = conn.CreateCommand())
{
cmd.CommandText = "INSERT INTO messages (message_id, conversation_id, tenant_id, direction, channel, " +
"content, delivery_status, created_at) VALUES (@m, @c, @t, 0, 1, " +
"'{\"blocks\": [{\"text\": \"legacy\", \"type\": 0, \"$type\": \"text\"}]}'::jsonb, 0, now())";
cmd.Parameters.Add(new NpgsqlParameter("m", messageId.Value));
cmd.Parameters.Add(new NpgsqlParameter("c", EntityId.New().Value));
cmd.Parameters.Add(new NpgsqlParameter("t", _tenant.Value));
await cmd.ExecuteNonQueryAsync();
}

var read = await _sut.GetByIdAsync(_tenant, messageId, CancellationToken.None);

read.Should().NotBeNull();
read!.Content.Blocks.Should().ContainSingle().Which.Should().BeOfType<TextBlock>()
.Which.Text.Should().Be("legacy");
}
}
Loading