redb.Route.Firebase 4.7.2

redb.Route.Firebase

Firebase transport for the redb.Route ESB framework — Firestore, Cloud Storage (GCS), and FCM in a single package.

Firestore producer with CRUD, queries, and 500-doc batch writes. Realtime snapshot listener consumer. Cloud Storage producer with upload/download (buffered or true streaming)/delete/copy/list/metadata, signed download links and bucket operations (create/delete/list, auto-create on the consumer). Polling consumer with idempotency (in-memory or a named shared repository), glob filtering, move/delete after read and a MoveFailed quarantine. FCM producer with full platform support — Android priority/TTL/channel, APNS content-available/mutable-content, WebPush.

Shared IFirebaseCredentialProvider across all three services — one credential setup, three transports.

Quick Start

services.AddRedbRoute(route =>
{
    route.Services.AddRedbRouteFirebase();

    // Realtime Firestore → log changes
    route.AddRoute("firestore-watch", r => r
        .From(Firestore.Collection("orders")
            .Where("status==pending")
            .CredentialPath("/secrets/firebase.json"))
        .To("log:order-change"));

    // Upload file to GCS bucket
    route.AddRoute("storage-upload", r => r
        .From("direct:upload")
        .To(FirebaseStorage.Bucket("media-uploads")
            .Operation(FirebaseStorageOperationType.Upload)
            .ObjectName(Expression.Header("fileName"))
            .ContentType("application/pdf")
            .CacheControl("public, max-age=86400")
            .CredentialPath("/secrets/firebase.json")));

    // Send push notification
    route.AddRoute("push-notify", r => r
        .From("direct:notify")
        .To(Fcm.Token(Expression.Header("deviceToken"))
            .Title("New Order")
            .Body(Expression.Simple("Order ${body['id']} is ready"))
            .CredentialPath("/secrets/firebase.json")));
});

Credential Resolution

All three components share the same resolution order:

1. ConnectionFactory — named IFirebaseCredentialProvider from the service registry
2. CredentialPath — path to a service-account JSON file
3. GOOGLE_APPLICATION_CREDENTIALS environment variable (default credential)
4. Emulator environment variables (for development) — FIRESTORE_EMULATOR_HOST (e.g. localhost:8086) and STORAGE_EMULATOR_HOST. The storage SDK uses STORAGE_EMULATOR_HOST verbatim as the service base URI, so for fake-gcs-server pass the full path: http://localhost:4443/storage/v1/.

// Option 1: Explicit credential file
.CredentialPath("/secrets/firebase.json")

// Option 2: Named provider from the context registry — several service accounts per process
context.AddToRegistry("myFirebase", new FirebaseCredentialProvider
{
    DefaultProjectId = "my-project",
    DefaultCredentialPath = "/secrets/firebase-sa.json",
});
// ...then on any endpoint:
.ConnectionFactory("myFirebase")

// Option 3: Environment variable (no code needed)
// Set GOOGLE_APPLICATION_CREDENTIALS=/secrets/firebase.json

Firebase apps are created as named instances — the connector never touches (or deletes) the process-global [DEFAULT] app, so it coexists with a host that initializes Firebase Admin SDK itself. The Firestore project id must be configured explicitly (endpoint projectId, AddRedbRouteFirebase(o => o.ProjectId = ...), DefaultProjectId, or the FIREBASE_PROJECT environment variable) — a missing project is a loud startup error, never a silent default.

Firestore (fstore)

URI Format

fstore://collection-path?option=value&...

Consumer — realtime snapshot listener (default):

fstore://orders?where=status==pending&credentialPath=/secrets/fb.json

Producer — default operation Set:

fstore://users?documentId=${header.userId}&merge=true
fstore://events?operation=Query&where=type==purchase;amount>100&orderBy=createdAt desc&limit=50
fstore://bulk?operation=BatchWrite

Sub-collections use path segments:

fstore://users/uid-123/orders?operation=Query&limit=10

Producer Operations

6 operations available, set via operation URI parameter.

Set with Merge

// Overwrite entire document (default)
route.From("direct:save")
    .To(Firestore.Collection("users")
        .DocumentId(Expression.Header("userId")));

// Merge — only update provided fields
route.From("direct:update-partial")
    .To(Firestore.Collection("users")
        .DocumentId(Expression.Header("userId"))
        .Merge());

Where syntax

field OP value conditions separated by ;. Operators: ==, !=, <, <=, >, >=, and array-contains (as a spaced token: tags array-contains 'admin'). Unquoted values are type-inferred (int/long/double/bool, else string); a single-quoted value is STRICTLY a string — status=='007' matches the string "007", never the number 7. ${...} templates in Where resolve per exchange on the producer Query operation only; a consumer with ${...} in Where fails at startup (the subscription query is built once — there is no exchange to resolve against).

Query

route.From("timer:poll?delay=60000")
    .To(Firestore.Collection("events")
        .Operation(FirestoreOperationType.Query)
        .Where("type==purchase;amount>100")
        .OrderBy("createdAt desc")
        .Limit(50))
    .To("log:results");

Filter syntax: field==value, separated by ;. Operators: ==, !=, >, >=, <, <=, array-contains.

BatchWrite

route.From("direct:bulk-import")
    .To(Firestore.Collection("products")
        .Operation(FirestoreOperationType.BatchWrite));

Automatically chunks into batches of 500 (Firestore hard limit). Body must be IEnumerable<IDictionary<string, object?>>.

With .DocumentIdField("id") the document id is taken from that field of each item (and the field itself is not written to the document); without it every item gets an auto-generated id. Note: batches commit sequentially — a failure between chunks leaves the earlier chunks written (the .NET SDK ships no BulkWriter, so cross-batch atomicity is not promised).

Consumer

Realtime snapshot listener — receives document changes as they happen via gRPC stream.

route.From(Firestore.Collection("orders")
        .Where("status==pending")
        .OrderBy("createdAt"))
    .Process(async (exchange, ct) =>
    {
        var changeType = exchange.In.GetHeader<string>(FirestoreHeaders.ChangeType);
        var docId = exchange.In.GetHeader<string>(FirestoreHeaders.DocumentId);
        // changeType: "Added", "Modified", or "Removed"
    });

Each document change creates a separate exchange. All changes within a snapshot are processed in parallel with graceful drain on stop.

A permanently failed listener (revoked credentials, deleted project) does not kill the consumer silently: the error is recorded in endpoint statistics and the subscription is re-created with exponential backoff (1s → 60s cap). Note that every (re)subscription delivers the current query results as an initial snapshot of Added changes. MaxConcurrency (default 16) bounds how many changes of one snapshot are processed in parallel.

Polling mode

.Realtime(false) switches to an honest poll loop: the query runs every Delay ms (InitialDelay before the first poll) and the result is diffed against the previous poll by DocumentId → UpdateTime — new id = Added, changed UpdateTime = Modified, id gone from the result = Removed (body is null; the old data is not retained). The first poll delivers everything as Added; the diff lives in memory, so a restart starts from a clean snapshot. Removed is only visible within the query result window — Where/Limit shift it.

Consumer Headers

Body Format

Default: Dictionary<string, object?> (Firestore native types). With RawJson(true): serialized JSON string.

Cloud Storage (fbstorage)

URI Format

fbstorage://bucket-name?option=value&...
fbstorage://bucket-name/prefix?option=value&...

The path after the bucket is a folder-like prefix: bucket/uploads and bucket/uploads/ both mean the uploads/ folder — Upload joins names with / (uploads/file.txt) and the consumer lists under the same folder. A raw (non-folder) string prefix is available via the prefix option instead.

Consumer — polls bucket for objects:

fbstorage://incoming-data?prefix=uploads/&include=*.csv&deleteAfterRead=true&delay=30000

Producer — default operation Upload:

fbstorage://media-bucket?objectName=${header.fileName}&contentType=image/png
fbstorage://media-bucket?operation=Download
fbstorage://media-bucket?operation=List&prefix=archive/

Producer Operations

5 operations available, set via operation URI parameter.

Upload

route.From("direct:upload")
    .To(FirebaseStorage.Bucket("media")
        .ObjectName(Expression.Simple("uploads/${header.category}/${header.fileName}"))
        .ContentType("application/pdf")
        .CacheControl("public, max-age=3600"));

Body can be byte[], Stream, string, or any object (auto-serialized to JSON). If ObjectName is not set, a GUID is generated.

Download with Streaming

// Default: byte[] in memory
route.From("direct:download")
    .To(FirebaseStorage.Bucket("media")
        .Operation(FirebaseStorageOperationType.Download));

// Large files: Stream (caller must dispose)
route.From("direct:download-large")
    .To(FirebaseStorage.Bucket("media")
        .Operation(FirebaseStorageOperationType.Download)
        .StreamBody());

Download runs object retrieval and metadata fetch in parallel for optimal performance.

Consumer

Polls the bucket with ListObjectsAsync, downloads matching objects, and optionally deletes/moves them.

route.From(FirebaseStorage.Bucket("incoming")
        .Prefix("data/")
        .Include("*.json")
        .Exclude("*.tmp")
        .DeleteAfterRead()
        .MaxMessagesPerPoll(50)
        .Delay(10_000)
        .CredentialPath("/secrets/firebase.json"))
    .To("direct:process");

Consumer Pipeline

ListObjects → Filter (prefix/include/exclude) → Idempotency Check → Download Body → Process → Post-Process (delete/move)

Consumer Options

Both prefixes take the file keys of the object (${file:ext}, ${file:name.noext}, ${file:parent}, ${file:size}, ${file:modified} and the rest of the file language), matching ignoring case. The object keeps its name: the move writes the prefix, keys substituted, followed by the object name, so moveFailed=failed/${file:ext}/ moves in/a.csv to failed/csv/in/a.csv. Renaming the object is not what these options do. ${file:name} is the whole object name, as in S3's doneFileName; the path keys write nothing. An unknown key fails when the endpoint is created.

Delete/move-after-read happens only when the exchange succeeded (no unhandled exception). A failed object stays in place and is retried on the next poll — or, with MoveFailed set, is moved to the quarantine prefix so the poll loop stops retrying it. With Idempotent enabled, only successfully processed objects are marked as seen.

Consumer Headers

FCM (fcm)

URI Format

fcm://send?messageType=Token&token=DEVICE_TOKEN&credentialPath=/secrets/fb.json
fcm://send?messageType=Topic&topic=news
fcm://send?messageType=Condition&condition='news' in topics && 'premium' in topics

Producer (send-only)

FCM is producer-only — no consumer. Sends push notifications via Firebase Cloud Messaging.

Beyond the classic send, three operations (?operation=): Multicast — the same message to many device tokens (IEnumerable<string> body or a comma-separated redbFcm.Tokens header); SubscribeToTopic / UnsubscribeFromTopic — topic management for a token list. All three report redbFcm.SuccessCount / redbFcm.FailureCount.

.To(Fcm.Multicast().Title("Broadcast").Build())            // body: List<string> of tokens
.To(Fcm.SubscribeToTopic("news").Build())                  // body: tokens to subscribe

Token Targeting (single device)

route.From("direct:push")
    .To(Fcm.Token(Expression.Header("deviceToken"))
        .Title("Order Shipped")
        .Body(Expression.Simple("Your order ${body['orderId']} is on the way"))
        .ImageUrl("https://example.com/shipped.png")
        .CredentialPath("/secrets/firebase.json"));

Topic Targeting (broadcast)

route.From("direct:broadcast")
    .To(Fcm.Topic("breaking-news")
        .Title("Breaking News")
        .Body(Expression.Simple("${body['headline']}"))
        .CredentialPath("/secrets/firebase.json"));

Condition Targeting (topic expressions)

route.From("direct:targeted")
    .To(Fcm.Condition("'premium' in topics && 'news' in topics")
        .Title("Premium News")
        .CredentialPath("/secrets/firebase.json"));

Data-Only Messages

// Silent push — no visible notification, just data payload
route.From("direct:sync")
    .To(Fcm.Token(Expression.Header("deviceToken"))
        .DataOnly()
        .CredentialPath("/secrets/firebase.json"));
// Body as Dictionary<string, string> → FCM data payload
// Or set headers with prefix: redbFcm.Data.key = value

Platform-Specific Configuration

route.From("direct:push")
    .To(Fcm.Token(Expression.Header("deviceToken"))
        .Title("Update Available")

        // Android
        .AndroidPriority("high")
        .AndroidTtlSeconds(3600)
        .AndroidChannelId("updates")

        // iOS
        .ApnsPriority("10")
        .ApnsCollapseId("update-group")

        // Web
        .WebPushLink("https://app.example.com/updates")

        .CredentialPath("/secrets/firebase.json"));

Dynamic Targeting via Headers

Target can be overridden at runtime by setting exchange headers:

exchange.In.Headers[FcmHeaders.Token] = "dynamic-device-token";
exchange.In.Headers[FcmHeaders.Title] = "Dynamic Title";
exchange.In.Headers[FcmHeaders.Body] = "Dynamic body text";

// Data payload via header prefix
exchange.In.Headers["redbFcm.Data.orderId"] = "12345";
exchange.In.Headers["redbFcm.Data.action"] = "refresh";

Output Headers

Fluent DSL Reference

using redb.Route.Firebase.Fluent;

Firestore

Cloud Storage

FCM

All builders support implicit operator string — pass directly to .From() / .To().

Part of

redb.Route — ESB & EIP Framework for .NET

Tracing

On the redb.Route activity source (AddSource("redb.Route")), with redb.route.endpoint:

• Consumers. Firestore, realtime and polling alike: one Consumer span per routed change, firestore {collection} receive. Storage: one per routed object, gcs {bucket} receive. A change or an object carries no trace context, so the span is a root, never a child of the activity the consumer inherited from whoever started the routes; nothing arriving opens none. A failed route marks it an error, whether the failure stays on the exchange or escapes it; our own stop does not. A Storage object whose route failed is not deleted either way.
• Producers. FCM (fcm {operation}), Firestore (firestore {operation}) and Storage (gcs {operation}) open one span per call; an error when the call fails, unless our own token cancelled it.
• RouteEngineOptions.EnableTelemetry = false opens none of these spans.