feat(queue): let a batch record the external order it fulfils

POST /queue/batches accepts external_source + external_ref; both are
returned on every batch and filterable on GET /queue/batches. The pair is
unique (index uq_print_batches_external), so a retried create answers 409
instead of queueing the same order twice. Migration covers SQLite and
PostgreSQL.

backend/app/api/routes/print_queue.py

@@ -9,6 +9,7 @@ from pathlib import Path
 import defusedxml.ElementTree as ET
 from fastapi import APIRouter, Depends, HTTPException, Query
 from sqlalchemy import and_, func, inspect, or_, select, update
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
@@ -1424,6 +1425,12 @@ async def _load_batch_for_write(
     return batch
 
 
+# Deliberately without the existing batch's id: the caller may not be allowed
+# to read it. They can look it up by the pair, which applies the usual
+# ownership rules.
+_EXTERNAL_REF_TAKEN = "A batch for this external_source and external_ref already exists"
+
+
 @router.post("/batches", response_model=PrintBatchResponse)
 async def create_batch(
     data: PrintBatchCreate,
@@ -1449,6 +1456,15 @@ async def create_batch(
 
     plate_targets = _validate_plate_targets(data.plates)
     await _validate_batch_project(db, data.project_id, current_user)
+    if data.external_source is not None:
+        existing = await db.execute(
+            select(PrintBatch.id).where(
+                PrintBatch.external_source == data.external_source,
+                PrintBatch.external_ref == data.external_ref,
+            )
+        )
+        if existing.scalar_one_or_none() is not None:
+            raise HTTPException(409, _EXTERNAL_REF_TAKEN)
 
     batch = PrintBatch(
         name=data.name.strip()[:255],
@@ -1460,9 +1476,17 @@ async def create_batch(
         project_id=data.project_id,
         due_date=data.due_date,
         notes=data.notes,
+        external_source=data.external_source,
+        external_ref=data.external_ref,
     )
     db.add(batch)
-    await db.flush()  # Need batch.id before assigning to items
+    try:
+        await db.flush()  # Need batch.id before assigning to items
+    except IntegrityError:
+        # Lost a race with a concurrent create for the same external record:
+        # the unique index caught what the lookup above could not.
+        await db.rollback()
+        raise HTTPException(409, _EXTERNAL_REF_TAKEN) from None
 
     if plate_targets is not None:
         for target in plate_targets:
@@ -1663,6 +1687,8 @@ async def ungroup_batch(
 @router.get("/batches", response_model=list[PrintBatchResponse])
 async def list_batches(
     status: str | None = Query(None, description="Filter by status (active, completed, cancelled)"),
+    external_source: str | None = Query(None, description="Filter by the integration that created the batch"),
+    external_ref: str | None = Query(None, description="Filter by the external record the batch fulfils"),
     db: AsyncSession = Depends(get_db),
     auth_result: tuple[User | None, bool] = Depends(
         require_ownership_permission(
@@ -1690,6 +1716,10 @@ async def list_batches(
     )
     if status:
         query = query.where(PrintBatch.status == status)
+    if external_source is not None:
+        query = query.where(PrintBatch.external_source == external_source)
+    if external_ref is not None:
+        query = query.where(PrintBatch.external_ref == external_ref)
     if current_user is not None and not can_read_all:
         query = query.where(PrintBatch.created_by_id == current_user.id)
     result = await db.execute(query)
@@ -1808,6 +1838,8 @@ async def _build_batch_response(
         project_id=batch.project_id,
         due_date=batch.due_date,
         notes=batch.notes,
+        external_source=batch.external_source,
+        external_ref=batch.external_ref,
         pending_count=progress.pending,
         printing_count=progress.printing,
         completed_count=progress.completed,

backend/app/core/database.py

@@ -4994,6 +4994,16 @@ async def run_migrations(conn):
     # Spoolman and the location sync then imported as storage locations.
     await _migrate_drop_ams_slot_locations(conn)
 
+    # Migration: link a batch to the external record that asked for it (a shop
+    # order an integration turned into prints). The unique index is what makes
+    # a retried create safe; both columns are new, so no row can violate it.
+    await _safe_execute(conn, "ALTER TABLE print_batches ADD COLUMN external_source VARCHAR(32)")
+    await _safe_execute(conn, "ALTER TABLE print_batches ADD COLUMN external_ref VARCHAR(255)")
+    await _safe_execute(
+        conn,
+        "CREATE UNIQUE INDEX IF NOT EXISTS uq_print_batches_external ON print_batches (external_source, external_ref)",
+    )
+
 
 async def _migrate_rename_ha_sensor_alert_template(conn) -> None:
     """Rename the ha_sensor_alert template to "Printer Sensor Alert" (#2824).

backend/app/models/print_batch.py

@@ -1,6 +1,6 @@
 from datetime import datetime
 
-from sqlalchemy import DateTime, ForeignKey, Integer, String, Text, UniqueConstraint, func
+from sqlalchemy import DateTime, ForeignKey, Index, Integer, String, Text, UniqueConstraint, func
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from backend.app.core.database import Base
@@ -21,6 +21,7 @@ class PrintBatch(Base):
     """
 
     __tablename__ = "print_batches"
+    __table_args__ = (Index("uq_print_batches_external", "external_source", "external_ref", unique=True),)
 
     id: Mapped[int] = mapped_column(primary_key=True)
     name: Mapped[str] = mapped_column(String(255))
@@ -45,6 +46,14 @@ class PrintBatch(Base):
     due_date: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
     notes: Mapped[str | None] = mapped_column(Text, nullable=True)
 
+    # Link to the record in another system that asked for this batch, e.g. a
+    # shop order an integration turned into prints. ``external_ref`` identifies
+    # this batch within ``external_source`` and is unique there, so a client
+    # that retries a create after a lost response gets a 409 instead of a
+    # second batch printing the same order twice. Both set or both null.
+    external_source: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    external_ref: Mapped[str | None] = mapped_column(String(255), nullable=True)
+
     # Timestamps
     created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
     completed_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)

backend/app/schemas/print_queue.py

@@ -389,6 +389,19 @@ class PrintBatchCreate(BaseModel):
     project_id: int | None = None
     due_date: datetime | None = None
     notes: str | None = None
+    # The external record this batch fulfils, for integrations. ``external_ref``
+    # is unique within ``external_source``: creating a second batch with the
+    # same pair is a 409, which makes a retried create safe.
+    external_source: str | None = Field(default=None, min_length=1, max_length=32, pattern=r"^[a-z0-9_-]+$")
+    external_ref: str | None = Field(default=None, min_length=1, max_length=255)
+
+    @model_validator(mode="after")
+    def _external_link_is_complete(self) -> "PrintBatchCreate":
+        # A ref without its source can't be looked up, and a source without a
+        # ref would escape the uniqueness guarantee (NULLs never collide).
+        if (self.external_source is None) != (self.external_ref is None):
+            raise ValueError("external_source and external_ref must be given together")
+        return self
 
 
 class PrintBatchUpdate(BaseModel):
@@ -465,6 +478,8 @@ class PrintBatchResponse(BaseModel):
     project_id: int | None = None
     due_date: UTCDatetime | None = None
     notes: str | None = None
+    external_source: str | None = None
+    external_ref: str | None = None
     # Derived counts
     pending_count: int = 0
     printing_count: int = 0

backend/tests/integration/test_print_batch_orders.py

@@ -1028,3 +1028,74 @@ class TestOrderSourcePreservation:
         assert deleted.json()["deleted"] is True
         db_session.expire_all()
         assert await db_session.get(PrintQueueItem, stray["id"]) is None
+
+
+class TestBatchExternalLink:
+    """A batch can carry the external record (e.g. a shop order) it fulfils."""
+
+    async def test_link_round_trips_and_filters(self, async_client, archive_factory):
+        archive = await archive_factory()
+        order = await _create_order(
+            async_client,
+            archive.id,
+            [{"plate_id": 1, "quantity_target": 2}],
+            external_source="shopify",
+            external_ref="shop.example/orders/1042/file/7",
+        )
+        assert order["external_source"] == "shopify"
+        assert order["external_ref"] == "shop.example/orders/1042/file/7"
+        await _create_order(async_client, archive.id, [{"plate_id": 1, "quantity_target": 1}])
+
+        by_source = (await async_client.get("/api/v1/queue/batches", params={"external_source": "shopify"})).json()
+        assert [b["id"] for b in by_source] == [order["id"]]
+        by_ref = (
+            await async_client.get(
+                "/api/v1/queue/batches",
+                params={"external_source": "shopify", "external_ref": "shop.example/orders/1042/file/7"},
+            )
+        ).json()
+        assert [b["id"] for b in by_ref] == [order["id"]]
+
+    async def test_second_create_for_the_same_record_is_refused(self, async_client, archive_factory):
+        """A retried create must not produce a second batch printing the order twice."""
+        archive = await archive_factory()
+        link = {"external_source": "shopify", "external_ref": "shop.example/orders/1042/file/7"}
+        await _create_order(async_client, archive.id, [{"plate_id": 1, "quantity_target": 1}], **link)
+
+        response = await async_client.post(
+            "/api/v1/queue/batches",
+            json={"name": "Retry", "archive_id": archive.id, "plates": [{"plate_id": 1, "quantity_target": 1}], **link},
+        )
+        assert response.status_code == 409
+        listed = (await async_client.get("/api/v1/queue/batches", params=link)).json()
+        assert len(listed) == 1
+
+    async def test_same_ref_under_another_source_is_a_different_record(self, async_client, archive_factory):
+        archive = await archive_factory()
+        plates = [{"plate_id": 1, "quantity_target": 1}]
+        await _create_order(async_client, archive.id, plates, external_source="shopify", external_ref="1042")
+        await _create_order(async_client, archive.id, plates, external_source="etsy", external_ref="1042")
+
+    async def test_unlinked_batches_never_collide(self, async_client, archive_factory):
+        """NULL pairs are exempt from the unique index."""
+        archive = await archive_factory()
+        plates = [{"plate_id": 1, "quantity_target": 1}]
+        await _create_order(async_client, archive.id, plates)
+        await _create_order(async_client, archive.id, plates)
+
+    @pytest.mark.parametrize(
+        "link",
+        [
+            {"external_source": "shopify"},
+            {"external_ref": "1042"},
+            {"external_source": "Shop ify", "external_ref": "1042"},
+            {"external_source": "shopify", "external_ref": ""},
+        ],
+    )
+    async def test_incomplete_or_malformed_link_is_rejected(self, async_client, archive_factory, link):
+        archive = await archive_factory()
+        response = await async_client.post(
+            "/api/v1/queue/batches",
+            json={"name": "Order", "archive_id": archive.id, "plates": [{"plate_id": 1, "quantity_target": 1}], **link},
+        )
+        assert response.status_code == 422

frontend/src/__tests__/components/BatchOrdersView.test.tsx

@@ -55,6 +55,8 @@ const batch = (over: Partial<PrintBatch> = {}): PrintBatch => {
   project_id: null,
   due_date: null,
   notes: null,
+  external_source: null,
+  external_ref: null,
   pending_count: 0,
   printing_count: 0,
   completed_count: 0,

frontend/src/api/client.ts

@@ -2658,6 +2658,9 @@ export interface PrintBatch {
   project_id: number | null;
   due_date: string | null;
   notes: string | null;
+  /** Set when an integration (e.g. a shop connector) created the batch. */
+  external_source: string | null;
+  external_ref: string | null;
   pending_count: number;
   printing_count: number;
   completed_count: number;