Messaging and Queueing

Summary

Zeeq uses a PostgreSQL-backed message queue for background work. The goal is to keep HTTP requests fast while giving longer-running work a reliable place to run out-of-band.

In practice, this means webhook ingestion, retryable jobs, scheduled work, and other background tasks can be handed off to a worker instead of blocking the web request that triggered them.

Zeeq uses Brighter for the command processor, message pump, handler pipeline, retry behavior, and PostgreSQL transport.

Why this layer exists

Brighter is a flexible library, but most feature code should not need to know about Brighter publications, subscriptions, queue tables, or routing keys. Those details are important, but they are platform concerns rather than product concerns.

Zeeq adds a small platform layer over Brighter so feature code can express intent: this message can be published, this handler consumes that message, this message belongs to a tenant, and this work is default, priority, or low-priority.

The runtime turns those declarations into concrete Brighter and PostgreSQL registrations. This keeps feature code easier to read and gives the platform one place to control queue layout, retry behavior, dead-letter storage, and worker configuration.

Key pieces

Publish flow

At a high level, feature code creates a message and calls IZeeqMessagePublisher.PublishAsync. Zeeq reads the logical topic from ConfigurePublisher, resolves the tenant tier and stable bucket for tenant messages, and then asks Brighter to write the message to the configured PostgreSQL queue table.

System messages skip tenant routing and use a fixed system route.

A worker later reads the message and dispatches it to the matching ZeeqMessageHandler<T>. If handling fails repeatedly, the shared retry pipeline gives up and Zeeq writes the message to the dead-letter table.

The feature code does not choose a queue table or build the final routing key (handled by internal platform routing).

Tenant routing

Tenant routing has two separate ideas that are easy to conflate.

Tenant tier decides where an organization's work is routed.

Message priority decides the default processing settings for that class of work.

For tenant messages, Zeeq resolves the organization's tier and hashes the organization id into a stable bucket inside that tier.

reports.refresh + org_123 + Default tier
  -> stable bucket 03 inside the default tier
  -> reports.refresh.default.03

This gives each organization a predictable route while still spreading organizations across queues. Priority tenants can get more capacity, low-tier tenants can get less capacity, and one noisy tenant is less likely to consume all worker capacity for the tier.

Worker modes

Zeeq can run consumers in two shapes.

The main switches are:

Configuration

The main configuration section is ZeeqMessaging.

PriorityDefaults also exists in code and is keyed by approved priority marker types such as PriorityMessage, DefaultMessage, and LowPriorityMessage. Use PriorityDefaults when an entire class of work should run faster or slower. Use TopicOverrides when one topic needs different behavior from the rest of its priority class.

PostgreSQL-specific settings live under ZeeqMessaging:Postgres.

Google Cloud Pub/Sub is selected with ZeeqMessaging:Provider = GcpPubSub. Postgres remains the default provider when the setting is omitted or set to Postgres.

{
  "ZeeqMessaging": {
    "Provider": "GcpPubSub",
    "GcpPubSub": {
      "ProjectId": "zeeq-dev-local",
      "MissingChannelPolicy": "Validate",
      "SubscriptionMode": "Stream",
      "UseEmulatorDetection": true
    }
  }
}

In local Aspire runs, the Pub/Sub emulator exports PUBSUB_EMULATOR_HOST and PUBSUB_PROJECT_ID to both zeeq-server and the split-mode zeeq-worker. If ProjectId is not set in configuration, the runtime falls back to PUBSUB_PROJECT_ID and then GCP_PROJECT_ID.

Use MissingChannelPolicy = Validate for steady-state local and production startup. Validation makes the app fail fast when a topic or subscription is missing, but it does not try to mutate Pub/Sub topology during normal process startup.

Use MissingChannelPolicy = Create only for explicit topology provisioning or exploratory emulator tests. Brighter's GCP Pub/Sub provider creates missing topics/subscriptions, but an already-provisioned emulator can reject Brighter's topic update path, and production runtime mutation would require broader IAM than steady-state publishing and subscribing need. Runtime services should be granted publisher/subscriber permissions for normal operation; topology creation should be handled by deployment/bootstrap tooling.

Topic retention is intentionally not enabled. This accepts the low criticality message-loss tradeoff during initial provisioning or cutover rather than paying the operational complexity and cost of retained undelivered messages.

Immediate GitHub acknowledgements

GitHub webhooks produce a few user-visible acknowledgement tasks that should not wait behind normal workflow work. Zeeq routes these through ImmediateMessage, which uses the shared messaging.brighter_messages_immediate queue table instead of tenant tier/bucket routing.

Current immediate GitHub topics:

The two topics have different safety models. Comment writes mutate a Zeeq-owned Markdown document and must use the comment lease so concurrent workers do not overwrite sections. Reaction writes are best-effort acknowledgement calls to GitHub's reaction API. They do not need a Zeeq dedupe table because GitHub treats repeated reactions by the same app/user as idempotent: a new reaction returns 201, and an already-present reaction returns 200.

For reactions, validation failures such as GitHub 422 are logged and acknowledged as no-op. Transient provider failures still flow through the normal message retry and dead-letter policy.

Common scenarios

Slow down local polling

Use this when local development does not need low-latency background work and you want fewer idle database polls.

{
  "ZeeqMessaging": {
    "Defaults": {
      // Empty queues are checked less often, which lowers idle database traffic.
      "PollIntervalMilliseconds": 5000
    }
  }
}

Give one topic a longer timeout

Use this when a specific handler normally takes longer than the default visibility timeout.

{
  "ZeeqMessaging": {
    "TopicOverrides": {
      // This is the logical topic from ConfigurePublisher, not the final queue route.
      "reports.refresh": {
        // Pull one expensive report refresh at a time.
        "BufferSize": 1,
        // Keep the message invisible long enough for the handler to finish.
        "VisibleTimeoutSeconds": 600
      }
    }
  }
}

Change tenant bucket allocation

Use this when the relative capacity between organization tiers changes.

{
  "ZeeqMessaging": {
    "TenantBuckets": {
      // More buckets means more independent routes for tenants in that tier.
      "PriorityBucketCount": 8,
      "DefaultBucketCount": 8,
      // Low-tier tenants get fewer routes and therefore less total capacity.
      "LowBucketCount": 4
    }
  }
}

Changing bucket counts changes route names, so treat bucket-count changes as queue migrations.

Declaring messages

A tenant message implements ITenantMessage and declares its logical topic.

// The generic priority marker selects the default processing profile.
[ConfigurePublisher<DefaultMessage>("reports.refresh")]
public sealed record RefreshReportMessage(
    // Tenant messages must expose OrganizationId so Zeeq can resolve tier and bucket.
    string OrganizationId,
    string? TeamId,
    string ReportId
) : Event(Id.Random()), ITenantMessage;

A consumer handler declares the message type and channel.

// The channel is the logical consumer identity for this handler.
[ConfigureConsumer<RefreshReportMessage>("reports.refresh.worker")]
public sealed class RefreshReportHandler(
    IDeadLetterWriter deadLetterWriter
) : ZeeqMessageHandler<RefreshReportMessage>(deadLetterWriter)
{
    protected override Task<RefreshReportMessage> HandleMessageAsync(
        RefreshReportMessage message,
        CancellationToken cancellationToken
    )
    {
        // Feature work goes here. Retry and dead-letter behavior live in the base class.
        return Task.FromResult(message);
    }
}

Feature code should publish through IZeeqMessagePublisher.

await publisher.PublishAsync(
    // The publisher resolves the concrete route; feature code only creates the message.
    new RefreshReportMessage(organizationId, teamId, reportId),
    cancellationToken
);

Failure behavior

Handlers use a shared retry pipeline through ZeeqMessageHandler<T>. If retries are exhausted, Zeeq writes the failed message to the PostgreSQL dead-letter table.

The dead-letter table is for inspection and future replay tooling. It is not another active Brighter queue.

Observability

The messaging layer participates in the same OpenTelemetry setup as the rest of the runtime. Zeeq registers the Brighter activity source and Npgsql instrumentation so publish, poll, acknowledge, and dead-letter database work can show up in traces.

Brighter also writes W3C trace context into the serialized queue message. That means a consumer can continue the trace from the request or background workflow that published the message instead of appearing as unrelated work.

Zeeq intentionally drops Brighter's root receive spans from export. Those spans are created before each Postgres poll and are mostly empty-queue timer noise. Message publish, process, handler, acknowledge/delete, and dead-letter spans are still recorded.

In development, Zeeq exposes a local-only diagnostic endpoint for a quick end-to-end check:

# Publishes a system diagnostic message and returns the smoke test id.
curl -X POST "http://zeeq-api.localhost:8095/api/v1/diagnostics/message-queue/smoke-test?note=local-check"

The server logs should show the same SmokeTestId on the publish and consume entries. When Aspire dashboard telemetry is available, aspire otel traces and aspire otel spans can be used to inspect the same flow.

References

• Brighter documentation
• Brighter GitHub repository
• Brighter external bus overview
• Brighter routing
• Brighter PostgreSQL message broker