Integrate Backend Services for AI Solutions

1. Queues vs Topics/Subscriptions

2. Message Properties

1. MessageId — unique identifier. Used for duplicate detection (idempotent receive).
2. SessionId — groups related messages. Required for session-based FIFO ordering. All messages with same SessionId go to same consumer.
3. ScheduledEnqueueTimeUtc — delay message delivery to a future time.
4. TimeToLive (TTL) — auto-move to DLQ if not consumed before TTL expires.
5. ApplicationProperties — custom key-value metadata for filtering subscriptions.

3. Receive Modes — The Exam Trap

1. Peek-lock (default, recommended) — message is locked for a timeout (default 60 seconds). Process succeeds → call complete(). Process fails → lock expires, message returns to queue for retry. Guarantees at-least-once delivery.
2. Receive-and-Delete — message deleted immediately on receive. Fast but if processing fails, message is lost forever. Only for non-critical, idempotent operations.

1. Send Messages

from azure.servicebus import ServiceBusClient, ServiceBusMessage
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
client = ServiceBusClient("mynamespace.servicebus.windows.net", credential)

with client.get_queue_sender("document-processing") as sender:
    msg = ServiceBusMessage(
        body='{"document_id": "doc-456", "url": "https://..."}',
        message_id="doc-456",                # For duplicate detection
        application_properties={"model": "gpt-4o", "priority": "high"}
    )
    sender.send_messages(msg)

2. Receive Messages (Peek-Lock)

with client.get_queue_receiver("document-processing", max_wait_time=5) as receiver:
    for msg in receiver:
        try:
            process_document(str(msg))       # Your business logic
            receiver.complete_message(msg)   # Removes from queue
        except Exception as e:
            receiver.abandon_message(msg)    # Returns to queue for retry
            # Or: receiver.dead_letter_message(msg, reason=str(e))  # Send to DLQ

3. Send Large Payloads — Claim-Check Pattern

Service Bus max message size = 256 KB (Standard) or 100 MB (Premium with large message support). AI models, embeddings, and documents often exceed this.

1. Upload large content to Azure Blob Storage
2. Send only the blob URI in the Service Bus message body
3. Receiver reads URI, fetches blob, processes large content

# Producer (send only the reference)
blob_uri = upload_to_blob(large_document_bytes)
sender.send_messages(ServiceBusMessage(
    body='{"blob_uri": "' + blob_uri + '", "document_id": "doc-123"}'
))

# Consumer (fetch the actual content)
msg_data = json.loads(str(msg))
content = download_from_blob(msg_data["blob_uri"])

1. Dead-Letter Queue (DLQ)

Sub-queue automatically created for every queue/subscription. Messages land in DLQ when:

1. Delivery count exceeds maximum (default 10 retries)
2. Consumer explicitly calls dead_letter_message()
3. Message TTL expires without being consumed
4. Filter evaluation fails on a topic subscription

Access the DLQ path: queuename/$deadletterqueue. Re-process or investigate failed messages there.

2. Sessions — FIFO Ordering Per Key

Standard Service Bus is not ordered. Sessions guarantee FIFO ordering for messages with the same session_id. Essential for conversation chains in AI chatbots where messages from the same user must be processed in order.

msg = ServiceBusMessage(body="step-2", session_id="user-001")
sender.send_messages(msg)

# Receiver acquires exclusive lock on a session
with receiver.get_session_receiver("my-queue", session_id="user-001") as session_receiver:
    for msg in session_receiver:
        process_in_order(msg)

3. Topic Filters — Route Messages to Subscriptions

1. Boolean filter — TrueRuleFilter(): all messages. FalseRuleFilter(): no messages.
2. SQL filter — SQL-like expression on ApplicationProperties: priority = 'high' AND model = 'gpt-4o'
3. Correlation filter — match specific property values exactly. More efficient than SQL filters.

1. HTTP Trigger

import azure.functions as func
import json

app = func.FunctionApp()

@app.route(route="embed", methods=["POST"])
def embed_text(req: func.HttpRequest) -> func.HttpResponse:
    body = req.get_json()
    embedding = generate_embedding(body.get("text", ""))
    return func.HttpResponse(
        json.dumps({"embedding": embedding}),
        mimetype="application/json"
    )

2. Service Bus Trigger

@app.service_bus_queue_trigger(
    arg_name="msg",
    queue_name="embed-queue",
    connection="ServiceBusConnection"
)
def process_embedding(msg: func.ServiceBusMessage) -> None:
    body = json.loads(msg.get_body().decode())
    embedding = generate_embedding(body["text"])
    store_in_cosmos(body["doc_id"], embedding)
    # Auto-completed on success; dead-lettered on repeated failure

3. Timer Trigger (NCRONTAB)

@app.timer_trigger(
    arg_name="timer",
    schedule="0 0 2 * * *"
)
def daily_cleanup(timer: func.TimerRequest) -> None:
    cleanup_old_embeddings()
    # NCRONTAB: {second} {minute} {hour} {day} {month} {dow}

4. Output Binding — No SDK Boilerplate

@app.route(route="process", methods=["POST"])
@app.cosmos_db_output(
    arg_name="doc",
    database_name="ai-db",
    container_name="results",
    connection="CosmosConnection"
)
def process_and_store(
    req: func.HttpRequest,
    doc: func.Out[func.Document]
) -> func.HttpResponse:
    result = process(req.get_json())
    doc.set(func.Document.from_dict(result))
    return func.HttpResponse("Stored", status_code=202)

import azure.durable_functions as df

@app.orchestration_trigger(context_name="context")
def embedding_orchestrator(context: df.DurableOrchestrationContext):
    docs = context.get_input()
    # Fan-out: embed all docs in parallel
    tasks = [context.call_activity("EmbedDoc", d) for d in docs]
    # Fan-in: wait for all
    embeddings = yield context.task_all(tasks)
    yield context.call_activity("StoreAll", embeddings)
    return f"Done: {len(embeddings)} docs"

@app.activity_trigger(input_name="doc")
def embed_doc(doc: dict) -> list:
    return generate_embedding(doc["content"])