fix(pipelines): seed standard processing pipeline as system default on startup

The pipeline management UI showed an empty list after first boot because no default system pipeline was created. This adds seed_default_pipeline() which: - Creates a system-owned (owner_id=NULL), is_default=True pipeline named "Standard Processing Pipeline" at application startup - Steps mirror the current hardcoded Celery processing workflow: convert_to_pdf → check_duplicates → ocr → extract_metadata → embed_metadata → compute_embedding → send_to_destinations - Is idempotent: no-op if any system pipeline already exists - Handles missing pipelines table gracefully (during first migration run) Also wires the seeder into app/main.py lifespan startup using the same pattern as seed_default_plans. 9 new tests added covering creation, step order, idempotency, and API visibility. Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
2026-03-07 10:58:39 +00:00
parent 1203a4b75f
commit d318110bbe
3 changed files with 223 additions and 0 deletions
@@ -792,3 +792,95 @@ def _unset_default(db: Session, owner_id: str | None) -> None:
        db.query(Pipeline).filter(Pipeline.owner_id == owner_id, Pipeline.is_default.is_(True)).update(
            {"is_default": False}
        )
+
+
+# ---------------------------------------------------------------------------
+# Default system pipeline seeding
+# ---------------------------------------------------------------------------
+
+# The steps that make up the standard document-processing workflow.  The order
+# here mirrors what the existing Celery-based pipeline executes for every
+# uploaded file.
+_DEFAULT_PIPELINE_STEPS: list[tuple[str, str]] = [
+    ("convert_to_pdf", "Convert to PDF"),
+    ("check_duplicates", "Check for Duplicates"),
+    ("ocr", "OCR Processing"),
+    ("extract_metadata", "Extract Metadata"),
+    ("embed_metadata", "Embed Metadata into PDF"),
+    ("compute_embedding", "Compute Text Embedding"),
+    ("send_to_destinations", "Send to Storage Destinations"),
+]
+
+#: Human-readable name shown in the management UI for the auto-seeded pipeline.
+DEFAULT_PIPELINE_NAME = "Standard Processing Pipeline"
+
+
+def seed_default_pipeline(db: Session) -> int:
+    """Ensure a system-owned default pipeline exists in the database.
+
+    This function is idempotent — it is a no-op when any system pipeline
+    (``owner_id IS NULL``) already exists.  It is intended to be called once
+    at application startup (in ``app.main.lifespan``) so that the pipeline
+    management UI always shows the default workflow that mirrors the existing
+    Celery-based processing steps.
+
+    The created pipeline:
+
+    * ``owner_id = None`` — owned by the system, visible to all users
+    * ``is_default = True`` — selected automatically for new documents
+    * Steps (in order): convert_to_pdf → check_duplicates → ocr →
+      extract_metadata → embed_metadata → compute_embedding →
+      send_to_destinations
+
+    Args:
+        db: An active SQLAlchemy session.
+
+    Returns:
+        ``1`` if a new pipeline was created, ``0`` if one already existed.
+    """
+    try:
+        if db.query(Pipeline).filter(Pipeline.owner_id.is_(None)).count() > 0:
+            return 0
+    except Exception:
+        # Table may not exist yet during the very first migration run.
+        return 0
+
+    pipeline = Pipeline(
+        owner_id=None,
+        name=DEFAULT_PIPELINE_NAME,
+        description=(
+            "The standard document processing workflow: PDF conversion, "
+            "duplicate detection, OCR, metadata extraction and embedding, "
+            "semantic embeddings, and final distribution to storage destinations."
+        ),
+        is_default=True,
+        is_active=True,
+    )
+    db.add(pipeline)
+    try:
+        db.flush()  # Assign pipeline.id without committing yet
+    except Exception as exc:  # pragma: no cover
+        db.rollback()
+        logger.error(f"Failed to create default pipeline: {exc}")
+        return 0
+
+    for pos, (step_type, label) in enumerate(_DEFAULT_PIPELINE_STEPS):
+        db.add(
+            PipelineStep(
+                pipeline_id=pipeline.id,
+                position=pos,
+                step_type=step_type,
+                label=label,
+                enabled=True,
+            )
+        )
+
+    try:
+        db.commit()
+        logger.info("Seeded default system pipeline: '%s' (id=%d)", DEFAULT_PIPELINE_NAME, pipeline.id)
+    except Exception as exc:  # pragma: no cover
+        db.rollback()
+        logger.error(f"Failed to seed default pipeline steps: {exc}")
+        return 0
+
+    return 1
@@ -112,6 +112,20 @@ async def lifespan(app: FastAPI):
    except Exception:
        logging.debug("Subscription plan seeding skipped — DB may not be ready yet")  # noqa: S110

+    # Seed the default system pipeline (mirrors the current hardcoded processing
+    # workflow) so it is immediately visible in the Pipelines management UI.
+    try:
+        from app.api.pipelines import seed_default_pipeline as _seed_pipeline
+        from app.database import SessionLocal as _SessionLocal  # noqa: F811 (re-import for clarity)
+
+        _db_pipeline = _SessionLocal()
+        try:
+            _seed_pipeline(_db_pipeline)
+        finally:
+            _db_pipeline.close()
+    except Exception:
+        logging.debug("Default pipeline seeding skipped — DB may not be ready yet")  # noqa: S110
+
    # Application is now running
    yield

@@ -465,3 +465,120 @@ class TestPipelineAPIHelpers:

        p = Pipeline(owner_id="user99", name="Private", is_default=False, is_active=True)
        assert _can_access_pipeline(p, "admin", admin=True) is True
+
+
+# ---------------------------------------------------------------------------
+# Unit + integration tests – seed_default_pipeline
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+class TestSeedDefaultPipeline:
+    """Tests for the seed_default_pipeline startup helper."""
+
+    def test_seed_creates_pipeline(self, db_session):
+        """seed_default_pipeline creates exactly one system pipeline."""
+        from app.api.pipelines import seed_default_pipeline
+
+        result = seed_default_pipeline(db_session)
+
+        assert result == 1
+        pipelines = db_session.query(Pipeline).filter(Pipeline.owner_id.is_(None)).all()
+        assert len(pipelines) == 1
+
+    def test_seeded_pipeline_is_default(self, db_session):
+        """The seeded pipeline has is_default=True and is_active=True."""
+        from app.api.pipelines import seed_default_pipeline
+
+        seed_default_pipeline(db_session)
+        p = db_session.query(Pipeline).filter(Pipeline.owner_id.is_(None)).first()
+
+        assert p is not None
+        assert p.is_default is True
+        assert p.is_active is True
+
+    def test_seeded_pipeline_has_correct_name(self, db_session):
+        """The seeded pipeline uses the canonical DEFAULT_PIPELINE_NAME."""
+        from app.api.pipelines import DEFAULT_PIPELINE_NAME, seed_default_pipeline
+
+        seed_default_pipeline(db_session)
+        p = db_session.query(Pipeline).filter(Pipeline.owner_id.is_(None)).first()
+
+        assert p.name == DEFAULT_PIPELINE_NAME
+
+    def test_seeded_pipeline_steps_count(self, db_session):
+        """The seeded pipeline has the correct number of steps."""
+        from app.api.pipelines import _DEFAULT_PIPELINE_STEPS, seed_default_pipeline
+
+        seed_default_pipeline(db_session)
+        p = db_session.query(Pipeline).filter(Pipeline.owner_id.is_(None)).first()
+        steps = db_session.query(PipelineStep).filter(PipelineStep.pipeline_id == p.id).all()
+
+        assert len(steps) == len(_DEFAULT_PIPELINE_STEPS)
+
+    def test_seeded_pipeline_step_types_and_order(self, db_session):
+        """Steps are in the correct order and match the expected step types."""
+        from app.api.pipelines import _DEFAULT_PIPELINE_STEPS, seed_default_pipeline
+
+        seed_default_pipeline(db_session)
+        p = db_session.query(Pipeline).filter(Pipeline.owner_id.is_(None)).first()
+        steps = (
+            db_session.query(PipelineStep)
+            .filter(PipelineStep.pipeline_id == p.id)
+            .order_by(PipelineStep.position)
+            .all()
+        )
+
+        expected_types = [step_type for step_type, _ in _DEFAULT_PIPELINE_STEPS]
+        actual_types = [s.step_type for s in steps]
+        assert actual_types == expected_types
+
+    def test_seeded_pipeline_all_steps_enabled(self, db_session):
+        """All seeded steps are enabled by default."""
+        from app.api.pipelines import seed_default_pipeline
+
+        seed_default_pipeline(db_session)
+        p = db_session.query(Pipeline).filter(Pipeline.owner_id.is_(None)).first()
+        steps = db_session.query(PipelineStep).filter(PipelineStep.pipeline_id == p.id).all()
+
+        assert all(s.enabled for s in steps), "All seeded steps should be enabled"
+
+    def test_seed_is_idempotent(self, db_session):
+        """Calling seed_default_pipeline twice does not create a duplicate."""
+        from app.api.pipelines import seed_default_pipeline
+
+        first = seed_default_pipeline(db_session)
+        second = seed_default_pipeline(db_session)
+
+        assert first == 1
+        assert second == 0  # No-op on second call
+
+        count = db_session.query(Pipeline).filter(Pipeline.owner_id.is_(None)).count()
+        assert count == 1
+
+    def test_seeded_pipeline_visible_via_api(self, client):
+        """The default pipeline is visible in the GET /api/pipelines listing."""
+        from app.api.pipelines import DEFAULT_PIPELINE_NAME, seed_default_pipeline
+        from app.database import get_db
+
+        # Seed using the same DB session that the test client uses
+        db = next(client.app.dependency_overrides[get_db]())
+        seed_default_pipeline(db)
+
+        r = client.get("/api/pipelines")
+        assert r.status_code == 200
+        names = [p["name"] for p in r.json()]
+        assert DEFAULT_PIPELINE_NAME in names
+
+    def test_seeded_pipeline_default_flag_visible_via_api(self, client):
+        """The seeded pipeline is returned with is_default=True via the API."""
+        from app.api.pipelines import DEFAULT_PIPELINE_NAME, seed_default_pipeline
+        from app.database import get_db
+
+        db = next(client.app.dependency_overrides[get_db]())
+        seed_default_pipeline(db)
+
+        r = client.get("/api/pipelines")
+        default_pipelines = [p for p in r.json() if p["name"] == DEFAULT_PIPELINE_NAME]
+        assert len(default_pipelines) == 1
+        assert default_pipelines[0]["is_default"] is True