Basis.IntegrationPlatform.MessageStore 0.3.0

Basis .NET Message Store Adapter

Implements message-store contract v1.0.0-rc2 (basis/integration-platform-contract).

NuGet package source for .NET integrations that need the Basis platform message-store contract.

Package (NuGet.org):

Basis.IntegrationPlatform.MessageStore

The adapter owns platform mechanics for .NET integrations:

• consumes the platform input topic APP_KAFKA_INPUT_TOPIC (default bip.<APP_NAME>.in)
• calls application code through IPlatformMessageHandler
• takes the platform attempt number from the x-bip-attempt header (absent means attempt 1) and upserts the (message_id, attempt) row so infrastructure redeliveries re-record the same attempt
• records terminal status in bip_message_log and bip_message_attempt
• records per-attempt step observations (DELIVERED/FAILED/SKIPPED) in bip_message_attempt.steps_json
• records idempotent external step outcomes in bip_step_outcome
• stores replayable payload bytes in Postgres under the payload policy (replayable = payload captured, not expired, not deleted — independent of message status)
• retries handler failures before terminal COMPLETED or DLQ
• publishes failed messages to APP_KAFKA_DLQ_TOPIC (default bip.message.dlq) with x-bip-error and x-bip-dlq-at
• exposes POST /admin/messages/{messageId}/resend
• republishes replay payloads back to the integration input topic (bip.<integration>.in), committing the REQUESTED audit row before the Kafka produce
• verifies the provisioned contract major in bip_contract at startup and refuses to start on a mismatch (missing table logs a loud warning during the v1.0 rollout)

Integration Usage

builder.Services.AddBasisMessageStoreAdapter<MyMessageHandler>();

app.MapBasisMessageStoreAdminEndpoints();

Application code implements:

public sealed class MyMessageHandler : IPlatformMessageHandler
{
    private readonly IPlatformStepExecutor steps;

    public MyMessageHandler(IPlatformStepExecutor steps)
    {
        this.steps = steps;
    }

    public async Task<PlatformMessageResult> HandleAsync(
        PlatformMessage message,
        CancellationToken cancellationToken)
    {
        await steps.ExecuteAsync(
            message,
            stepId: "target-system",
            endpointUri: "https://target.example/api/orders",
            targetRef: message.TargetSystem,
            operation: async ct =>
            {
                await SendToTargetSystem(message, ct);
            },
            cancellationToken);

        return PlatformMessageResult.Completed();
    }
}

Required Runtime Environment

The shared integration-app chart injects these when the catalog enables both Kafka and Postgres:

APP_NAME
APP_ENVIRONMENT
APP_TENANT
KAFKA_BOOTSTRAP_SERVERS
APP_KAFKA_INPUT_TOPIC
APP_KAFKA_DLQ_TOPIC
POSTGRES_JDBC_URL
POSTGRES_USERNAME
POSTGRES_PASSWORD
BASIS_MESSAGE_STORE_SINK
BASIS_MESSAGE_STORE_STEP_OUTCOME_STORE
BASIS_MESSAGE_STORE_REPLAY_INPUT_TOPIC_PATTERN

Payload policy

Identical semantics across runtime adapters (contract v1 §6):

metadata_only rows store no payload_bytes and are replayable=false with a reason. Replayability is a property of payload availability, never of the message outcome: COMPLETED messages are resendable as long as the payload is captured, unexpired, and undeleted.

Publish

Releases are published to NuGet.org as Basis.IntegrationPlatform.MessageStore by .github/workflows/publish-nuget.yml (tag-driven, NuGet Trusted Publishing). The workflow builds, tests against the vendored contract bip_* schema in tests/fixtures/postgres/, packs with the version derived from the tag, and pushes. Per ADR-003 the public registry is the canonical channel; tenant-registry fan-out remains an opt-in overlay for restricted environments and is operated from the platform's internal infrastructure.

License and support

Apache-2.0. The adapter is developed as part of the Basis integration platform; production use is supported for platform customers through their platform agreement. Issues and pull requests from other users are welcome and handled best-effort.