Tracking Delivery

After sending a document, track its journey through the Peppol network to confirm delivery.

Transaction Lifecycle

┌─────────┐   ┌────────────┐   ┌─────────┐   ┌───────────┐
│ pending │──▶│ processing │──▶│ sending │──▶│ delivered │
└─────────┘   └────────────┘   └─────────┘   └───────────┘
                                    │
                                    ▼
                              ┌──────────┐
                              │  failed  │
                              └──────────┘

Status	Description	Duration
pending	Queued for processing	Seconds
processing	Validation in progress	1-5 seconds
sending	Being transmitted to receiver	5-30 seconds
delivered	Successfully delivered	Terminal
failed	Delivery failed	Terminal

Checking Status

Get Transaction Details

import requests

def get_transaction_status(transaction_id: str) -> dict:
    """Get current status of a transaction."""
    response = requests.get(
        f"https://app.goroute.ai/peppol-api/api/v1/transactions/{transaction_id}",
        headers={"X-API-Key": "your_api_key"}
    )
    response.raise_for_status()
    return response.json()

# Example usage
status = get_transaction_status("txn_abc123")
print(f"Status: {status['status']}")
print(f"Created: {status['created_at']}")
print(f"Updated: {status['updated_at']}")

Response Structure

{
    "id": "txn_abc123",
    "status": "delivered",
    "document_type": "Invoice",
    "document_id": "INV-2024-00123",
    "sender": {
        "scheme": "0106",
        "identifier": "12345678",
        "name": "Sender Company BV"
    },
    "receiver": {
        "scheme": "0106",
        "identifier": "87654321",
        "name": "Receiver Company BV"
    },
    "created_at": "2024-01-15T10:30:00Z",
    "updated_at": "2024-01-15T10:30:45Z",
    "delivered_at": "2024-01-15T10:30:45Z",
    "message_id": "msg_xyz789",
    "metadata": {
        "reference_id": "your-internal-ref-123"
    }
}

Polling for Status

For asynchronous sends, poll until delivery is confirmed:

import time

def wait_for_delivery(transaction_id: str, timeout: int = 120) -> dict:
    """Wait for transaction to reach terminal status."""
    start_time = time.time()
    
    while time.time() - start_time < timeout:
        status = get_transaction_status(transaction_id)
        
        if status["status"] == "delivered":
            return status
        
        if status["status"] == "failed":
            raise Exception(f"Delivery failed: {status.get('error')}")
        
        # Wait before polling again
        time.sleep(2)
    
    raise TimeoutError(f"Transaction {transaction_id} did not complete within {timeout}s")

# Usage
result = wait_for_delivery("txn_abc123")
print(f"Delivered at: {result['delivered_at']}")

Webhooks (Recommended)

Instead of polling, receive status updates via webhooks:

Configure Webhook

# Register a webhook endpoint
response = requests.post(
    "https://app.goroute.ai/peppol-api/api/v1/webhooks",
    headers={
        "X-API-Key": "your_api_key",
        "Content-Type": "application/json"
    },
    json={
        "url": "https://your-app.com/webhooks/goroute",
        "events": ["transaction.delivered", "transaction.failed"],
        "secret": "your_webhook_secret"
    }
)

Webhook Events

Event	Description
transaction.created	Document accepted for processing
transaction.processing	Validation started
transaction.sending	Transmission to receiver started
transaction.delivered	Successfully delivered
transaction.failed	Delivery failed

Webhook Payload

{
    "id": "evt_abc123",
    "type": "transaction.delivered",
    "created_at": "2024-01-15T10:30:45Z",
    "data": {
        "transaction_id": "txn_abc123",
        "status": "delivered",
        "document_type": "Invoice",
        "document_id": "INV-2024-00123",
        "sender": {
            "scheme": "0106",
            "identifier": "12345678"
        },
        "receiver": {
            "scheme": "0106",
            "identifier": "87654321"
        },
        "delivered_at": "2024-01-15T10:30:45Z",
        "as4_message_id": "msg_xyz789"
    }
}

Handle Webhooks

from flask import Flask, request
import hmac
import hashlib

app = Flask(__name__)
WEBHOOK_SECRET = "your_webhook_secret"

@app.route("/webhooks/goroute", methods=["POST"])
def handle_webhook():
    # Verify signature
    signature = request.headers.get("X-GoRoute-Signature")
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        request.data,
        hashlib.sha256
    ).hexdigest()
    
    if not hmac.compare_digest(signature, expected):
        return {"error": "Invalid signature"}, 401
    
    event = request.json
    
    if event["type"] == "transaction.delivered":
        handle_delivery(event["data"])
    elif event["type"] == "transaction.failed":
        handle_failure(event["data"])
    
    return {"status": "ok"}

def handle_delivery(data):
    """Process successful delivery."""
    transaction_id = data["transaction_id"]
    document_id = data["document_id"]
    
    # Update your database
    db.update_invoice_status(document_id, "delivered")
    
    # Log the delivery
    logger.info(f"Invoice {document_id} delivered via {transaction_id}")

def handle_failure(data):
    """Process delivery failure."""
    transaction_id = data["transaction_id"]
    error = data.get("error", {})
    
    # Alert the team
    notify_team(f"Delivery failed: {transaction_id}", error)
    
    # Maybe queue for retry
    if error.get("code") in ["RECEIVER_UNAVAILABLE", "TIMEOUT"]:
        queue_for_retry(transaction_id)

List Transactions

Query your transaction history:

# List recent transactions
response = requests.get(
    "https://app.goroute.ai/peppol-api/api/v1/transactions",
    params={
        "status": "delivered",
        "limit": 100,
        "offset": 0
    },
    headers={"X-API-Key": "your_api_key"}
)

transactions = response.json()
print(f"Total: {transactions['total']}")
for txn in transactions["items"]:
    print(f"{txn['id']}: {txn['document_id']} - {txn['status']}")

Query Parameters

Parameter	Type	Description
status	string	Filter by status
document_type	string	Filter by document type
sender_id	string	Filter by sender identifier
receiver_id	string	Filter by receiver identifier
from_date	string	Start date (ISO 8601)
to_date	string	End date (ISO 8601)
limit	integer	Results per page (default: 50)
offset	integer	Pagination offset

Example: Find Failed Transactions

# Get all failed transactions in the last 24 hours
from datetime import datetime, timedelta

yesterday = (datetime.utcnow() - timedelta(days=1)).isoformat() + "Z"

response = requests.get(
    "https://app.goroute.ai/peppol-api/api/v1/transactions",
    params={
        "status": "failed",
        "from_date": yesterday
    },
    headers={"X-API-Key": "your_api_key"}
)

failed = response.json()["items"]
print(f"Failed in last 24h: {len(failed)}")

for txn in failed:
    print(f"  {txn['document_id']}: {txn['error']['message']}")

Delivery Receipt (MDN)

For delivered transactions, you can retrieve the AS4 Message Disposition Notification:

# Get delivery receipt
response = requests.get(
    f"https://app.goroute.ai/peppol-api/api/v1/transactions/{transaction_id}/receipt",
    headers={"X-API-Key": "your_api_key"}
)

receipt = response.json()
print(f"Receipt ID: {receipt['mdn_id']}")
print(f"Received At: {receipt['received_at']}")
print(f"Receiver AP: {receipt['receiver_access_point']}")

Receipt Structure

{
    "mdn_id": "mdn_abc123",
    "transaction_id": "txn_abc123",
    "message_id": "msg_xyz789",
    "received_at": "2024-01-15T10:30:45Z",
    "receiver_access_point": {
        "name": "Receiver AP",
        "peppol_id": "POP001234"
    },
    "signature_valid": true,
    "raw_mdn": "..."  // Optional: raw AS4 receipt XML
}

Monitoring Dashboard

Track transactions in the GoRoute dashboard:

1. Log in to app.goroute.ai
2. Navigate to Transactions
3. Filter by status, date range, or document type
4. Click a transaction for details

Best Practices

1. Use Webhooks Over Polling

# ❌ Don't do this in production
while True:
    status = get_transaction_status(txn_id)
    if status["status"] in ["delivered", "failed"]:
        break
    time.sleep(2)

# ✅ Use webhooks instead
@app.route("/webhooks/goroute", methods=["POST"])
def webhook():
    event = request.json
    process_event(event)

2. Store Transaction IDs

# Save transaction ID with your invoice
def send_and_track(invoice_xml: str, invoice_id: str):
    result = client.send_invoice(invoice_xml)
    
    # Store mapping
    db.save_transaction_mapping(
        invoice_id=invoice_id,
        transaction_id=result["transaction_id"]
    )
    
    return result

3. Handle Webhook Failures

# Implement idempotency
def handle_webhook(event):
    event_id = event["id"]
    
    # Check if already processed
    if db.event_exists(event_id):
        return {"status": "already_processed"}
    
    # Process event
    process_event(event)
    
    # Mark as processed
    db.mark_event_processed(event_id)
    
    return {"status": "ok"}

Next Steps

• Attachments
• Batch Sending
• Receiving Documents