feat: destination cards, DAG run log, notifications, source management, and UI fixes

- Pipeline: add destination card rendering with proper CSS styling
- DAG: add run log CRUD + pause/resume/stop/retry/delete endpoints
- Run Log: repository for DAG run log persistence
- Notifications: live toast with auto-dismiss + notification page
- Sources: table counts tracking, source list, create/edit forms
- Dashboard: pipeline-level metrics card
- Users: user management list
- UI: CSS style overhaul for all pages
- Clean: remove .dockerignore, remove .bak files

Author: edsuwarna

.dockerignore

@@ -0,0 +1,34 @@
+"""add table_count and enabled_table_count columns to sources table
+
+Revision ID: 011
+Revises: 010
+Create Date: 2026-06-16 10:45:00.000000
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision: str = '011'
+down_revision: Union[str, None] = '010'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        'sources',
+        sa.Column('table_count', sa.Integer(), nullable=False, server_default=sa.text('0')),
+        schema='arus_config'
+    )
+    op.add_column(
+        'sources',
+        sa.Column('enabled_table_count', sa.Integer(), nullable=False, server_default=sa.text('0')),
+        schema='arus_config'
+    )
+
+
+def downgrade() -> None:
+    op.drop_column('sources', 'table_count', schema='arus_config')
+    op.drop_column('sources', 'enabled_table_count', schema='arus_config')

alembic/versions/011_add_source_table_counts.py

@@ -10,9 +10,10 @@
 
 from arus.shared.db.session import get_db
 from arus.modules.auth.router import get_current_user
-from arus.modules.run_log.models import Run
+from arus.modules.run_log.models import Run, RunTableStat
 from arus.modules.pipeline.models import Pipeline, PipelineTable
 from arus.modules.source.models import Source
+from arus.modules.destination.models import Destination
 
 router = APIRouter(prefix="/api/dag", tags=["dag"])
 
@@ -28,19 +29,40 @@ async def get_dag_data(
 
     for p in pipelines:
         source = db.query(Source).filter(Source.id == p.source_id).first()
+        destination = db.query(Destination).filter(Destination.id == p.destination_id).first()
         tables = db.query(PipelineTable).filter(
             PipelineTable.pipeline_id == p.id, PipelineTable.enabled == True
         ).all()
 
-        # Get last run status per table
+        # Get last run and per-table stats
         last_run = (
             db.query(Run)
             .filter(Run.pipeline_id == str(p.id))
             .order_by(desc(Run.started_at))
             .first()
         )
-
         pipeline_status = last_run.status if last_run else "not_started"
+        currently_running = last_run and last_run.status == "running"
+
+        # Get per-table stats from the last completed run
+        table_stats = {}
+        if last_run:
+            stats = db.query(RunTableStat).filter(RunTableStat.run_id == last_run.id).all()
+            for s in stats:
+                table_stats[s.table_name] = s
+
+        def asset_status(table_name):
+            """Derive per-asset status. If currently running, all show 'running'.
+            Otherwise check per-table stat, fallback to pipeline-level status."""
+            if currently_running:
+                return "running"
+            stat = table_stats.get(table_name)
+            if stat:
+                if stat.error_message:
+                    return "failed"
+                return "success"
+            return pipeline_status
+
         safe_name = source.name.lower().replace("-", "_").replace(" ", "_") if source else "unknown"
 
         assets = []
@@ -49,40 +71,80 @@ async def get_dag_data(
 
         for t in tables:
             load_mode = getattr(t, "load_mode", None) or p.load_mode or "direct"
+            status = asset_status(t.source_table)
+            stat = table_stats.get(t.source_table)
 
-            # Source asset always exists
+            # ---- Source asset ----
             assets.append({
                 "name": t.source_table,
                 "layer": "source",
-                "status": pipeline_status,
+                "status": status,
                 "table": t.source_table,
+                "schema": t.source_schema,
+                "sync_mode": t.sync_mode,
+                "source_type": source.type if source else None,
+                "source_host": source.host if source else None,
+                "source_port": source.port if source else None,
+                "source_database": source.database if source else None,
+                "rows": stat.rows_extracted if stat else 0,
+                "duration_ms": stat.duration_ms if stat else 0,
+                "error": stat.error_message if stat else None,
             })
 
             if load_mode == "raw":
-                # Raw mode: source → raw → target
+                # ---- Raw mode: source → raw → target ----
                 has_raw = True
                 raw_name = f"{safe_name}_{t.source_table}_raw"
+                raw_status = asset_status(raw_name)
+                raw_stat = table_stats.get(raw_name)
+
                 assets.append({
                     "name": raw_name,
                     "layer": "raw",
-                    "status": pipeline_status,
+                    "status": raw_status,
                     "table": t.source_table,
+                    "schema": t.source_schema,
+                    "sync_mode": t.sync_mode,
+                    "raw_table": raw_name,
+                    "raw_schema": destination.raw_schema if destination else "staging",
+                    "rows": raw_stat.rows_loaded_raw if raw_stat else 0,
+                    "duration_ms": raw_stat.duration_ms if raw_stat else 0,
+                    "error": raw_stat.error_message if raw_stat else None,
                 })
+
+                target_status = asset_status(t.source_table)
+                target_stat = table_stats.get(t.source_table)
+
                 assets.append({
                     "name": t.source_table,
                     "layer": "target",
-                    "status": pipeline_status,
+                    "status": target_status,
                     "table": t.source_table,
+                    "schema": t.source_schema,
+                    "sync_mode": t.sync_mode,
+                    "target_table": t.source_table,
+                    "target_schema": t.target_schema or (destination.target_schema if destination else None),
+                    "rows": target_stat.rows_loaded_analytics if target_stat else 0,
+                    "duration_ms": target_stat.duration_ms if target_stat else 0,
+                    "error": target_stat.error_message if target_stat else None,
                 })
+
                 edges.append({"from": t.source_table, "to": raw_name})
                 edges.append({"from": raw_name, "to": t.source_table})
             else:
-                # Direct mode: source → target
+                # ---- Direct mode: source → target ----
                 assets.append({
                     "name": t.source_table,
                     "layer": "target",
-                    "status": pipeline_status,
+                    "status": status,
                     "table": t.source_table,
+                    "schema": t.source_schema,
+                    "sync_mode": t.sync_mode,
+                    "target_table": t.source_table,
+                    "target_schema": t.target_schema or (destination.target_schema if destination else None),
+                    "rows": stat.rows_loaded_analytics if stat else 0,
+                    "duration_ms": stat.duration_ms if stat else 0,
+                    "error": stat.error_message if stat else None,
                 })
                 edges.append({"from": t.source_table, "to": t.source_table})
 

arus/modules/dag/router.py

@@ -32,6 +32,39 @@ def get_runs(self, pipeline_id: str, limit: int = 20, offset: int = 0) -> list[R
             .all()
         )
 
+    def get_runs_by_asset(
+        self, pipeline_id: str, asset_name: str, limit: int = 20, offset: int = 0
+    ) -> list[dict]:
+        """Get runs for a pipeline, joined with RunTableStat filtered by asset_name.
+        Returns per-asset metrics instead of pipeline-level totals."""
+        rows = (
+            self.db.query(Run, RunTableStat)
+            .join(RunTableStat, RunTableStat.run_id == Run.id)
+            .filter(Run.pipeline_id == pipeline_id)
+            .filter(RunTableStat.table_name == asset_name)
+            .order_by(desc(Run.started_at))
+            .offset(offset)
+            .limit(limit)
+            .all()
+        )
+        return [
+            {
+                "id": str(run.Run.id),
+                "pipeline_id": str(run.Run.pipeline_id),
+                "status": run.Run.status,
+                "started_at": run.Run.started_at,
+                "finished_at": run.Run.finished_at,
+                "trigger_type": run.Run.trigger_type,
+                "rows_synced": (
+                    (run.RunTableStat.rows_loaded_analytics or 0)
+                    + (run.RunTableStat.rows_loaded_raw or 0)
+                ),
+                "duration_ms": run.RunTableStat.duration_ms or run.Run.duration_ms,
+                "error_message": run.RunTableStat.error_message or run.Run.error_message,
+            }
+            for run in rows
+        ]
+
     def get_recent_runs(self, limit: int = 10) -> list[Run]:
         return self.db.query(Run).order_by(desc(Run.started_at)).limit(limit).all()
 

arus/modules/run_log/repository.py

@@ -19,13 +19,15 @@ async def list_runs(
     pipeline_id: str,
     limit: int = Query(20, le=100),
     offset: int = Query(0, ge=0),
+    asset_name: str = Query(None, description="Filter runs by asset/table name"),
     repo: RunLogRepository = Depends(get_run_log_repo),
     user: dict = Depends(get_current_user),
 ):
-    runs = repo.get_runs(pipeline_id, limit=limit, offset=offset)
-    return {
-        "status": "ok",
-        "data": [
+    if asset_name:
+        runs = repo.get_runs_by_asset(pipeline_id, asset_name, limit=limit, offset=offset)
+    else:
+        rows = repo.get_runs(pipeline_id, limit=limit, offset=offset)
+        runs = [
             {
                 "id": str(r.id),
                 "pipeline_id": str(r.pipeline_id),
@@ -34,10 +36,14 @@ async def list_runs(
                 "finished_at": r.finished_at,
                 "duration_ms": r.duration_ms,
                 "trigger_type": r.trigger_type,
+                "rows_synced": r.rows_synced or 0,
                 "error_message": r.error_message,
             }
-            for r in runs
-        ],
+            for r in rows
+        ]
+    return {
+        "status": "ok",
+        "data": runs,
     }
 
 

arus/modules/run_log/router.py

@@ -24,6 +24,8 @@ class Source(Base):
     table_include = Column(ARRAY(String), default=[])
     table_exclude = Column(ARRAY(String), default=[])
     schema_include = Column(ARRAY(String), default=[])
+    table_count = Column(Integer, default=0)
+    enabled_table_count = Column(Integer, default=0)
     status = Column(String(20), default="registered")
     last_tested = Column(DateTime(timezone=True), nullable=True)
     created_at = Column(DateTime(timezone=True), server_default=func.now())

arus/modules/source/models.py

@@ -40,8 +40,8 @@ async def list_sources(
                     "uri": s.uri,
                     "auth_source": s.auth_source,
                     "status": s.status,
-                    "table_count": 0,
-                    "enabled_table_count": 0,
+                    "table_count": s.table_count or 0,
+                    "enabled_table_count": s.enabled_table_count or 0,
                     "schema_include": s.schema_include or [],
                     "last_tested": s.last_tested,
                     "created_at": s.created_at,
@@ -92,6 +92,8 @@ async def get_source(
             "table_include": source.table_include,
             "table_exclude": source.table_exclude,
             "schema_include": source.schema_include or [],
+            "table_count": source.table_count or 0,
+            "enabled_table_count": source.enabled_table_count or 0,
             "status": source.status,
             "last_tested": source.last_tested,
             "created_at": source.created_at,
@@ -178,4 +180,15 @@ async def update_source_tables(
 
     pipe_service = PipelineService(PipelineRepository(db), db)
     result = pipe_service.auto_create_from_source(source_id, str(default_dest.id), req.tables)
+
+    # Update source table counts after saving selection
+    source_repo = SourceRepository(db)
+    source = source_repo.get_by_id(source_id)
+    if source:
+        enabled = sum(1 for t in req.tables if t.get("enabled", True))
+        source.table_count = len(req.tables)
+        source.enabled_table_count = enabled
+        source.status = "connected"
+        db.commit()
+
     return {"status": "ok", "data": result}

arus/modules/source/router.py

@@ -2,6 +2,7 @@
 from arus.shared.crypto import decrypt_password
 from arus.modules.connector.registry import get_source
 from arus.modules.source.repository import SourceRepository
+from sqlalchemy import func
 
 
 class SourceService:
@@ -63,8 +64,15 @@ def discover_tables(self, source_id: str) -> list[dict]:
                     "watermark_column": sync.watermark_column,
                     "enabled": True,
                 })
+            # Update source metadata after successful discovery
+            source.table_count = len(tables)
+            source.last_tested = func.now()
+            source.status = "connected"
+            self.repo.db.commit()
             return result
         except Exception as e:
+            source.status = "error"
+            self.repo.db.commit()
             raise DiscoveryFailedError(f"Discovery failed: {str(e)}")
 
     def discover_schemas(self, source_id: str) -> list[str]: