fix: retain the production QStash token region

.github/workflows/build-images.yml

@@ -774,6 +774,7 @@ jobs:
           API_WEBHOOK_ENDPOINT: https://api.knowhereto.ai/v1/internal/s3-events
           SNS_TOPIC_ARN: arn:aws:sns:us-east-1:107424103509:knowhere-prod-s3-events
           QSTASH_CALLBACK_BASE_URL: https://api.knowhereto.ai/api/v1
+          QSTASH_BASE_URL: https://qstash-us-east-1.upstash.io
           API_CPU: "512"
           API_MEMORY: "2048"
           WORKER_CPU: "2048"

deploy/ecs/README.md

@@ -41,6 +41,14 @@ python deploy/ecs/render_task_definitions.py --environment staging --output-dir
 
 The output directory is deployment-only and must not be committed. The renderer fails on missing inputs, unresolved placeholders, or either long-lived S3 credential variable.
 
+`QSTASH_BASE_URL` is an optional renderer input applied to both containers.
+The production release workflow supplies `https://qstash-us-east-1.upstash.io`
+because the existing production QStash token belongs to `us-east-1`. Omitting
+this input preserves the SDK default for other environments. QStash API
+publication and the public callback base URL are separate configuration values;
+the API base URL must match the token region. A wrong region rejects publication
+before any customer webhook is delivered.
+
 The staging workflow keeps the API at 256 CPU / 1024 MiB. The production release
 workflow sets the API to 512 CPU / 2048 MiB (2 GiB) and the worker to 2048 CPU /
 4096 MiB, so production API memory remains at 2 GiB across future releases.

deploy/ecs/render_task_definitions.py

@@ -95,6 +95,20 @@ def render_template(template_path: Path, output_path: Path, variables: dict[str,
     """Render one JSON template to a deployment-only output path."""
     definition = json.loads(template_path.read_text(encoding="utf-8"))
     rendered = replace_placeholders(definition, variables)
+    qstash_base_url: str = variables.get("QSTASH_BASE_URL", "")
+    if qstash_base_url:
+        if not isinstance(rendered, dict):
+            raise ValueError("Task definition must be an object")
+        containers: object = rendered.get("containerDefinitions")
+        if not isinstance(containers, list):
+            raise ValueError("Task definition must contain containerDefinitions")
+        for container in containers:
+            if not isinstance(container, dict):
+                raise ValueError("Task container must be an object")
+            environment: object = container.get("environment")
+            if not isinstance(environment, list):
+                raise ValueError("Task container must contain environment entries")
+            environment.append({"name": "QSTASH_BASE_URL", "value": qstash_base_url})
     validate_rendered_definition(rendered)
     output_path.parent.mkdir(parents=True, exist_ok=True)
     output_path.write_text(json.dumps(rendered, indent=2) + "\n", encoding="utf-8")
@@ -119,6 +133,9 @@ def load_variables() -> dict[str, str]:
     missing = [name for name, value in variables.items() if not value]
     if missing:
         raise ValueError(f"Missing required renderer variables: {', '.join(missing)}")
+    # Preserve the SDK default when no override is supplied, while production
+    # explicitly selects the region that owns its existing QStash token.
+    variables["QSTASH_BASE_URL"] = os.environ.get("QSTASH_BASE_URL", "")
     return variables
 
 

deploy/ecs/test_render_task_definitions.py

@@ -96,6 +96,39 @@ def test_staging_templates_render_without_long_lived_s3_keys(
     assert "${" not in output_path.read_text(encoding="utf-8")
 
 
+@pytest.mark.parametrize(
+    "template_name",
+    ["task-definition-api.staging.json", "task-definition-worker.staging.json"],
+)
+def test_production_templates_publish_webhooks_in_the_token_region(
+    tmp_path: Path,
+    template_name: str,
+) -> None:
+    """Both production containers keep the approved QStash region on release."""
+    output_path: Path = tmp_path / template_name
+    render_template(
+        TEMPLATE_DIRECTORY / template_name,
+        output_path,
+        {
+            **RENDER_VARIABLES,
+            "DEPLOYMENT_ENVIRONMENT": "prod",
+            "RUNTIME_ENVIRONMENT": "production",
+            "APP_ENV": "production",
+            "QSTASH_BASE_URL": "https://qstash-us-east-1.upstash.io",
+        },
+    )
+    rendered: dict[str, object] = json.loads(output_path.read_text(encoding="utf-8"))
+    containers: list[dict[str, object]] = rendered["containerDefinitions"]
+    for container in containers:
+        environment: list[dict[str, str]] = container["environment"]
+        region_entries: list[dict[str, str]] = [
+            entry for entry in environment if entry["name"] == "QSTASH_BASE_URL"
+        ]
+        assert region_entries == [
+            {"name": "QSTASH_BASE_URL", "value": "https://qstash-us-east-1.upstash.io"}
+        ]
+
+
 @pytest.mark.parametrize(
     ("template_name", "container_name", "expected_environment"),
     [