id: shopify-stale-unfulfilled-paid-order-sentinel
namespace: company.ecommerce
inputs:
- id: shopify_store_domain
type: STRING
displayName: Shopify Store Domain
defaults: "your-store.myshopify.com"
description: Hostname of the target Shopify store domain.
- id: max_fulfillment_delay_hours
type: INT
displayName: Maximum Allowable Fulfillment Delay (Hours)
defaults: 48
description: Maximum permissible hours between paid order placement and shipping
fulfillment before triggering an escalation.
- id: order_batch_limit
type: INT
displayName: Order Batch Limit
defaults: 50
description: Maximum number of unfulfilled paid orders to inspect per audit run.
- id: max_tolerable_stale_orders
type: INT
displayName: Maximum Tolerable Overdue Orders
defaults: 0
description: Number of SLA-breaching unfulfilled orders acceptable before
triggering an alert.
triggers:
- id: weekday_fulfillment_audit_schedule
type: io.kestra.plugin.core.trigger.Schedule
description: Audits fulfillment queues twice daily on weekdays at 08:00 and
14:00 UTC. Shipped disabled by default.
cron: "0 8,14 * * 1-5"
disabled: true
tasks:
- id: fetch_unfulfilled_orders
type: io.kestra.plugin.graphql.Request
description: Query Shopify Admin GraphQL API for paid unfulfilled orders.
uri: "https://{{ inputs.shopify_store_domain }}/admin/api/2024-01/graphql.json"
headers:
X-Shopify-Access-Token: "{{ secret('SHOPIFY_ADMIN_ACCESS_TOKEN') }}"
Content-Type: "application/json"
User-Agent: "Kestra-Shopify-Fulfillment-Sentinel/1.0"
failOnGraphQLErrors: false
query: |
query ($query: String!, $limit: Int!) {
orders(first: $limit, query: $query, sortKey: CREATED_AT, reverse: false) {
nodes {
id
name
createdAt
displayFinancialStatus
displayFulfillmentStatus
totalPriceSet {
shopMoney {
amount
currencyCode
}
}
customer {
displayName
email
}
lineItems(first: 3) {
nodes {
title
quantity
}
}
}
}
}
variables:
query: "financial_status:paid fulfillment_status:unfulfilled"
limit: "{{ inputs.order_batch_limit }}"
- id: audit_fulfillment_sla
type: io.kestra.plugin.scripts.python.Script
description: Identify paid orders exceeding the fulfillment aging SLA and
compute revenue impact.
containerImage: python:3.11-slim
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
env:
RESPONSE_BODY: "{{ outputs.fetch_unfulfilled_orders.body }}"
MAX_DELAY_HOURS: "{{ inputs.max_fulfillment_delay_hours }}"
STORE_DOMAIN: "{{ inputs.shopify_store_domain }}"
script: |
import os
import json
from datetime import datetime, timezone
from kestra import Kestra
raw_data = os.environ.get("RESPONSE_BODY", "{}")
max_delay = int(os.environ.get("MAX_DELAY_HOURS", 48))
store_domain = os.environ.get("STORE_DOMAIN", "")
try:
data = json.loads(raw_data)
except Exception:
data = {}
orders = data.get("data", {}).get("orders", {}).get("nodes", [])
now = datetime.now(timezone.utc)
stale_orders = []
total_stale_revenue = 0.0
currency = "USD"
for order in orders:
created_at_str = order.get("createdAt")
if not created_at_str:
continue
try:
created_at = datetime.fromisoformat(created_at_str.replace("Z", "+00:00"))
age_hours = (now - created_at).total_seconds() / 3600.0
except Exception:
age_hours = 0.0
if age_hours >= max_delay:
money = order.get("totalPriceSet", {}).get("shopMoney", {})
amount = float(money.get("amount", 0.0))
currency = money.get("currencyCode", currency)
total_stale_revenue += amount
customer = order.get("customer") or {}
customer_name = customer.get("displayName") or customer.get("email") or "Anonymous"
stale_orders.append({
"id": order.get("id"),
"name": order.get("name"),
"age_hours": round(age_hours, 1),
"amount": amount,
"customer": customer_name
})
summary = f"Detected {len(stale_orders)} paid order(s) waiting fulfillment > {max_delay}h across {store_domain} representing {currency} {total_stale_revenue:,.2f} in delayed goods."
order_lines = [
f"• *{o['name']}* - {currency} {o['amount']:,.2f} (Customer: `{o['customer']}`, Age: `{o['age_hours']}h`)"
for o in stale_orders[:5]
]
Kestra.outputs({
"stale_orders_count": len(stale_orders),
"total_stale_revenue": round(total_stale_revenue, 2),
"currency": currency,
"summary": summary,
"orders_formatted": "\n".join(order_lines)
})
- id: check_sla_violations
type: io.kestra.plugin.core.flow.If
description: Evaluate if overdue order count breaches configured tolerance.
condition: "{{ outputs.audit_fulfillment_sla.vars.stale_orders_count >
inputs.max_tolerable_stale_orders }}"
then:
- id: notify_slack_fulfillment_alert
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Post unfulfilled order SLA breach alert to warehouse and logistics
channels in Slack.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{
"channel": "#fulfillment-logistics-alerts",
"username": "Kestra Shopify Fulfillment Sentinel",
"icon_emoji": ":truck:",
"attachments": [
{
"color": "#ff9900",
"title": "Shopify Paid Orders Fulfillment SLA Breach",
"title_link": "https://{{ inputs.shopify_store_domain }}/admin/orders?fulfillment_status=unfulfilled",
"text": "{{ outputs.audit_fulfillment_sla.vars.summary }}",
"fields": [
{
"title": "Store Domain",
"value": "`{{ inputs.shopify_store_domain }}`",
"short": true
},
{
"title": "SLA Threshold",
"value": "{{ inputs.max_fulfillment_delay_hours }} hours",
"short": true
},
{
"title": "Overdue Orders Count",
"value": "{{ outputs.audit_fulfillment_sla.vars.stale_orders_count }}",
"short": true
},
{
"title": "Delayed Revenue Impact",
"value": "{{ outputs.audit_fulfillment_sla.vars.currency }} {{ outputs.audit_fulfillment_sla.vars.total_stale_revenue }}",
"short": true
},
{
"title": "Oldest Unfulfilled Orders",
"value": "{{ outputs.audit_fulfillment_sla.vars.orders_formatted }}",
"short": false
}
],
"footer": "Kestra Fulfillment Watchdog",
"ts": "{{ now() | date('X') }}"
}
]
}
errors:
- id: send_failure_alert
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Dispatch critical alert if Shopify GraphQL query or fulfillment
audit fails.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{
"channel": "#devops-alerts",
"username": "Kestra Sentinel Watchdog",
"icon_emoji": ":rotating_light:",
"attachments": [
{
"color": "#dc3545",
"title": "Shopify Fulfillment Sentinel Execution Failed",
"text": "Execution *{{ execution.id }}* failed during fulfillment SLA audit for store *{{ inputs.shopify_store_domain }}*.",
"fields": [
{
"title": "Flow ID",
"value": "{{ flow.id }}",
"short": true
},
{
"title": "Namespace",
"value": "{{ flow.namespace }}",
"short": true
}
],
"footer": "Kestra Sentinel Watchdog",
"ts": "{{ now() | date('X') }}"
}
]
}
