diff --git a/api/elasticsearch_api.py b/api/elasticsearch_api.py index 4674628..d803ded 100644 --- a/api/elasticsearch_api.py +++ b/api/elasticsearch_api.py @@ -39,6 +39,11 @@ BUSINESS_TABLES = { ("mongodb_supplychain", "supplychain", "events"), ("cassandra_telemetry", "telemetry", "device_metrics"), } +# prefer rows that actually carry the business identifiers (names are sparse in raw tables) +BUSINESS_WHERE = { + ("postgres_sales", "public", "sales_orders"): "customer_name IS NOT NULL", + ("mysql_hr", "hr", "employee_events"): "employee_name IS NOT NULL", +} # de-duplicated business entities so users can search people/customers by name + id ENTITY_QUERIES = [ ( @@ -61,6 +66,41 @@ ENTITY_QUERIES = [ ), ] +# business index names (federated source data + entity rollups) +IDX_ORDERS = "atc-postgres-sales-public-sales-orders" +IDX_HR = "atc-mysql-hr-hr-employee-events" +IDX_SUPPLY = "atc-mongodb-supplychain-supplychain-events" +IDX_TELEMETRY = "atc-cassandra-telemetry-telemetry-device-metrics" +IDX_CUSTOMERS = "atc-customers" +IDX_EMPLOYEES = "atc-employees" +IDX_PRODUCTS = "atc-products" + +# lenient multi-format date parsing for the timestamp columns in the data +_DATE_FMT = ( + "yyyy-MM-dd HH:mm:ss.SSS zzz||yyyy-MM-dd HH:mm:ss zzz||" + "yyyy-MM-dd HH:mm:ss.SSS||yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||" + "strict_date_optional_time||epoch_millis" +) +_NUM = {"type": "double", "ignore_malformed": True} +_DATE = {"type": "date", "format": _DATE_FMT, "ignore_malformed": True} +# index template so measures are numeric (sum/avg) and timestamps are real dates +ATC_TEMPLATE = { + "index_patterns": [f"{INDEX_PREFIX}*"], + "priority": 200, + "template": { + "settings": {"number_of_replicas": 0, "refresh_interval": "5s"}, + "mappings": { + "properties": { + "amount": _NUM, "total_amount": _NUM, "salary_change": _NUM, + "unit_price": _NUM, "metric_value": _NUM, "quantity": _NUM, + "revenue": _NUM, "order_count": _NUM, + "order_ts": _DATE, "event_ts": _DATE, "metric_ts": _DATE, + "order_date": _DATE, "ts": _DATE, + } + }, + }, +} + router = APIRouter(prefix="/api/search", tags=["search"]) @@ -300,6 +340,154 @@ async def es_search(q: str = Query(..., min_length=1), size: int = Query(10, le= return await search_query({"q": q, "size": size}) +# ────────────────────────────────────────────────────────────────────────────── +# Business data overview (data abstraction — charts & filters over the real data) +# ────────────────────────────────────────────────────────────────────────────── +_biz_cache: dict[str, Any] = {} +_BIZ_TTL = 30.0 +_totals_cache: dict[str, Any] = {"at": 0.0, "data": {}} +_TOTALS_TTL = 600.0 + + +def _buckets(agg: dict, key: str, metric: str | None = None) -> list[dict]: + out = [] + for b in (agg.get(key, {}) or {}).get("buckets", []): + item = {"key": b.get("key_as_string", b.get("key")), "count": b.get("doc_count", 0)} + if metric and metric in b: + item["value"] = round(b[metric].get("value") or 0, 2) + out.append(item) + return out + + +async def _es_aggs(client: httpx.AsyncClient, index: str, filt: dict, aggs: dict) -> dict: + try: + r = await client.post( + f"{ELASTICSEARCH_URL}/{index}/_search", + params={"ignore_unavailable": "true", "allow_no_indices": "true"}, + auth=_auth(), + json={"size": 0, "query": filt, "aggs": aggs, "track_total_hits": True}, + ) + if r.status_code >= 400: + return {} + return r.json() + except Exception: # noqa: BLE001 + return {} + + +def _source_totals() -> dict: + """True row counts in the source systems via cheap planner estimates (cached).""" + if time.time() - _totals_cache["at"] < _TOTALS_TTL and _totals_cache["data"]: + return _totals_cache["data"] + out: dict[str, Any] = {} + try: + import sql_console as s + out["orders"] = s._table_row_count("postgres", "public.sales_orders") + out["hr_events"] = s._table_row_count("mysql", "hr.employee_events") + out["supply_events"] = s._table_row_count("mongodb", "supplychain.events") + except Exception: # noqa: BLE001 + pass + _totals_cache.update({"at": time.time(), "data": out}) + return out + + +@router.get("/business") +async def business_overview(q: str = Query(""), region: str = Query("")): + cache_key = f"{q}|{region}" + cached = _biz_cache.get(cache_key) + if cached and time.time() - cached["at"] < _BIZ_TTL: + return cached["data"] + + must: list[dict] = [] + if q.strip(): + must.append({"query_string": {"query": q, "lenient": True, "default_operator": "AND"}}) + region_must = must + ([{"term": {"region.keyword": region}}] if region.strip() else []) + filt = {"bool": {"must": must or [{"match_all": {}}]}} + rfilt = {"bool": {"must": region_must or [{"match_all": {}}]}} + + async with httpx.AsyncClient(timeout=20.0, verify=False) as client: + orders = await _es_aggs(client, IDX_ORDERS, rfilt, { + "by_region": {"terms": {"field": "region.keyword", "size": 12}, "aggs": {"rev": {"sum": {"field": "amount"}}}}, + "by_channel": {"terms": {"field": "sales_channel.keyword", "size": 12}, "aggs": {"rev": {"sum": {"field": "amount"}}}}, + "by_status": {"terms": {"field": "order_status.keyword", "size": 12}}, + "top_customers": {"terms": {"field": "customer_name.keyword", "size": 10}, "aggs": {"spend": {"sum": {"field": "amount"}}}}, + "top_products": {"terms": {"field": "product_id", "size": 10}, "aggs": {"rev": {"sum": {"field": "amount"}}}}, + "over_time": {"date_histogram": {"field": "order_ts", "calendar_interval": "month", "min_doc_count": 0}, "aggs": {"rev": {"sum": {"field": "amount"}}}}, + "revenue": {"sum": {"field": "amount"}}, + "avg_order": {"avg": {"field": "amount"}}, + }) + hr = await _es_aggs(client, IDX_HR, rfilt, { + "by_department": {"terms": {"field": "department.keyword", "size": 15}}, + "by_role": {"terms": {"field": "role_name.keyword", "size": 12}}, + "by_event": {"terms": {"field": "event_type.keyword", "size": 12}}, + }) + supply = await _es_aggs(client, IDX_SUPPLY, rfilt, { + "by_type": {"terms": {"field": "type.keyword", "size": 15}}, + "by_region": {"terms": {"field": "region.keyword", "size": 12}}, + }) + telemetry = await _es_aggs(client, IDX_TELEMETRY, filt, { + "by_metric": {"terms": {"field": "metric_type.keyword", "size": 12}, "aggs": {"avg": {"avg": {"field": "metric_value"}}}}, + }) + + async def _count(index: str, body_filt: dict) -> int: + try: + r = await client.post( + f"{ELASTICSEARCH_URL}/{index}/_count", + params={"ignore_unavailable": "true", "allow_no_indices": "true"}, + auth=_auth(), json={"query": body_filt}, + ) + return int(r.json().get("count", 0)) if r.status_code < 400 else 0 + except Exception: # noqa: BLE001 + return 0 + + customers = await _count(IDX_CUSTOMERS, rfilt) + employees = await _count(IDX_EMPLOYEES, rfilt) + products = await _count(IDX_PRODUCTS, rfilt) + + o_agg = orders.get("aggregations", {}) + hr_agg = hr.get("aggregations", {}) + sup_agg = supply.get("aggregations", {}) + tel_agg = telemetry.get("aggregations", {}) + orders_indexed = orders.get("hits", {}).get("total", {}).get("value", 0) + + data = { + "ok": True, + "generated_at": datetime.now(timezone.utc).isoformat(), + "filters": {"q": q, "region": region}, + "kpis": { + "customers": customers, + "employees": employees, + "products": products, + "orders_indexed": orders_indexed, + "revenue_indexed": round((o_agg.get("revenue", {}) or {}).get("value") or 0, 2), + "avg_order": round((o_agg.get("avg_order", {}) or {}).get("value") or 0, 2), + "telemetry_indexed": telemetry.get("hits", {}).get("total", {}).get("value", 0), + "supply_indexed": supply.get("hits", {}).get("total", {}).get("value", 0), + "source_totals": _source_totals(), + }, + "orders": { + "by_region": _buckets(o_agg, "by_region", "rev"), + "by_channel": _buckets(o_agg, "by_channel", "rev"), + "by_status": _buckets(o_agg, "by_status"), + "top_customers": _buckets(o_agg, "top_customers", "spend"), + "top_products": _buckets(o_agg, "top_products", "rev"), + "over_time": _buckets(o_agg, "over_time", "rev"), + }, + "hr": { + "by_department": _buckets(hr_agg, "by_department"), + "by_role": _buckets(hr_agg, "by_role"), + "by_event": _buckets(hr_agg, "by_event"), + }, + "supply": { + "by_type": _buckets(sup_agg, "by_type"), + "by_region": _buckets(sup_agg, "by_region"), + }, + "telemetry": {"by_metric": _buckets(tel_agg, "by_metric", "avg")}, + "regions": [b["key"] for b in _buckets(o_agg, "by_region")], + } + _biz_cache[cache_key] = {"at": time.time(), "data": data} + return data + + # ────────────────────────────────────────────────────────────────────────────── # Indexer (Trino-federated → ES atc-* indices) # ────────────────────────────────────────────────────────────────────────────── @@ -345,8 +533,9 @@ def _meta(catalog: str, schema: str, table: str, source: str) -> dict: } -def _index_trino_table(sqlmod, client: httpx.Client, catalog: str, schema: str, table: str, limit: int = ROWS_PER_TABLE) -> int: - res = sqlmod._run_trino(f'SELECT * FROM "{catalog}"."{schema}"."{table}" LIMIT {limit}', limit) +def _index_trino_table(sqlmod, client: httpx.Client, catalog: str, schema: str, table: str, limit: int = ROWS_PER_TABLE, where: str | None = None) -> int: + where_sql = f" WHERE {where}" if where else "" + res = sqlmod._run_trino(f'SELECT * FROM "{catalog}"."{schema}"."{table}"{where_sql} LIMIT {limit}', limit) if not res.get("ok"): raise RuntimeError(res.get("error", "query failed")[:200]) cols = res.get("columns", []) @@ -447,11 +636,39 @@ def _index_catalog(client: httpx.Client) -> int: return written +def _ensure_template(client: httpx.Client) -> None: + try: + client.put(f"{ELASTICSEARCH_URL}/_index_template/atc-data", json=ATC_TEMPLATE) + except Exception as exc: # noqa: BLE001 + _reindex_state["errors"].append(f"template: {str(exc)[:120]}") + + +def _drop_atc_indices(client: httpx.Client) -> None: + """Delete existing atc-* indices so they are re-created with the new + numeric/date mappings from the template (mappings can't be changed in place).""" + try: + r = client.get(f"{ELASTICSEARCH_URL}/_cat/indices/{INDEX_PREFIX}*?format=json&h=index") + for it in (r.json() if r.status_code < 400 else []): + nm = it.get("index") + if nm: + try: + client.delete(f"{ELASTICSEARCH_URL}/{nm}") + except Exception: # noqa: BLE001 + pass + except Exception as exc: # noqa: BLE001 + _reindex_state["errors"].append(f"cleanup: {str(exc)[:120]}") + + def _reindex_worker(): import sql_console as sqlmod try: client = _es_client(timeout=90.0) with client: + # numeric/date mappings + fresh indices + _reindex_state["current"] = "preparing mappings" + _ensure_template(client) + _drop_atc_indices(client) + # catalog/metadata _reindex_state["current"] = "catalog inventory" _reindex_state["total_docs"] += _index_catalog(client) @@ -484,7 +701,8 @@ def _reindex_worker(): _reindex_state["current"] = f"{catalog}.{schema}.{table}" try: cap = ROWS_BUSINESS if (catalog, schema, table) in BUSINESS_TABLES or catalog == "iceberg" else ROWS_PER_TABLE - n = _index_trino_table(sqlmod, client, catalog, schema, table, cap) + where = BUSINESS_WHERE.get((catalog, schema, table)) + n = _index_trino_table(sqlmod, client, catalog, schema, table, cap, where) idx = _sanitize_index(catalog, schema, table) _reindex_state["indices"][idx] = n _reindex_state["total_docs"] += n diff --git a/api/sql_console.py b/api/sql_console.py index 12695dd..ba36551 100644 --- a/api/sql_console.py +++ b/api/sql_console.py @@ -355,6 +355,12 @@ class RowUpdateRequest(BaseModel): changes: dict[str, Any] = Field(..., min_length=1) +class RowInsertRequest(BaseModel): + engine: str = Field(..., pattern="^(postgres|mysql|mongodb|cassandra|neo4j)$") + object: str = Field(..., min_length=1, max_length=256) + values: dict[str, Any] = Field(..., min_length=1) + + def _connection_info(engine: str) -> dict[str, str]: if engine == "postgres": return {"host": DB_HOST, "port": str(PG_PORT), "database": PG_DB, "user": PG_USER} @@ -1001,6 +1007,115 @@ def _update_row(engine: str, object_name: str, pk: dict[str, Any], changes: dict return {"ok": True, "engine": engine, "object": object_name, "statement": sql, "cdc": engine in ("postgres", "mysql", "mongodb")} +def _insert_postgres(object_name: str, values: dict[str, Any]) -> str: + schema, table = _parse_fqn("postgres", object_name) + cols = [c for c in values if values[c] is not None] + if not cols: + raise ValueError("No values provided") + for c in cols: + if not _safe_ident(c): + raise ValueError(f"Invalid column name: {c}") + col_sql = ", ".join(f'"{c}"' for c in cols) + ph = ", ".join(["%s"] * len(cols)) + sql = f'INSERT INTO "{schema}"."{table}" ({col_sql}) VALUES ({ph})' + conn = psycopg2.connect(host=DB_HOST, port=PG_PORT, user=PG_USER, password=PG_PASS, dbname=PG_DB, connect_timeout=8) + try: + conn.autocommit = True + conn.cursor().execute(sql, [values[c] for c in cols]) + return sql + finally: + conn.close() + + +def _insert_mysql(object_name: str, values: dict[str, Any]) -> str: + _schema, table = _parse_fqn("mysql", object_name) + cols = [c for c in values if values[c] is not None] + if not cols: + raise ValueError("No values provided") + for c in cols: + if not _safe_ident(c): + raise ValueError(f"Invalid column name: {c}") + col_sql = ", ".join(f"`{c}`" for c in cols) + ph = ", ".join(["%s"] * len(cols)) + sql = f"INSERT INTO `{table}` ({col_sql}) VALUES ({ph})" + conn = pymysql.connect(host=DB_HOST, port=MYSQL_PORT, user=MYSQL_USER, password=MYSQL_PASS, database=MYSQL_DB, connect_timeout=8) + try: + conn.autocommit = True + conn.cursor().execute(sql, [values[c] for c in cols]) + return sql + finally: + conn.close() + + +def _insert_mongodb(object_name: str, values: dict[str, Any]) -> str: + db_name, coll = _parse_fqn("mongodb", object_name) + doc = {k: v for k, v in values.items() if v is not None} + if not doc: + raise ValueError("No values provided") + client = _mongo_client() + try: + res = client[db_name][coll].insert_one(doc) + return f"INSERT {db_name}.{coll} _id={res.inserted_id}" + finally: + client.close() + + +def _insert_cassandra(object_name: str, values: dict[str, Any]) -> str: + ks, table = _parse_fqn("cassandra", object_name) + cols = [c for c in values if values[c] is not None] + if not cols: + raise ValueError("No values provided") + for c in cols: + if not _safe_ident(c): + raise ValueError(f"Invalid column name: {c}") + col_sql = ", ".join(cols) + ph = ", ".join(["%s"] * len(cols)) + cql = f"INSERT INTO {ks}.{table} ({col_sql}) VALUES ({ph})" + cluster = _cass_cluster() + session = cluster.connect() + try: + session.execute(cql, [values[c] for c in cols]) + return cql + finally: + cluster.shutdown() + + +def _insert_neo4j(object_name: str, values: dict[str, Any]) -> str: + label = object_name.split(".")[-1] + props = {k: _coerce_value(v) for k, v in values.items() if v is not None} + if not props: + raise ValueError("No values provided") + for c in props: + if not _safe_ident(c): + raise ValueError(f"Invalid property name: {c}") + set_frag = ", ".join(f"n.{c} = ${c}" for c in props) + cypher = f"CREATE (n:`{label}`) SET {set_frag} RETURN n" + driver = _neo4j_driver() + try: + with driver.session() as session: + session.run(cypher, **props).consume() + return cypher + finally: + driver.close() + + +def _insert_row(engine: str, object_name: str, values: dict[str, Any]) -> dict[str, Any]: + values = {k: _coerce_value(v) for k, v in values.items()} + if engine == "postgres": + sql = _insert_postgres(object_name, values) + elif engine == "mysql": + sql = _insert_mysql(object_name, values) + elif engine == "mongodb": + sql = _insert_mongodb(object_name, values) + elif engine == "cassandra": + sql = _insert_cassandra(object_name, values) + elif engine == "neo4j": + sql = _insert_neo4j(object_name, values) + else: + raise ValueError(f"Insert not supported for {engine}") + return {"ok": True, "engine": engine, "object": object_name, "statement": sql, "cdc": engine in ("postgres", "mysql", "mongodb")} + + @router.get("/samples/{engine}") async def get_samples(engine: str): if engine not in SAMPLES and engine not in LAKE_ENGINES: @@ -1130,6 +1245,16 @@ async def update_row(body: RowUpdateRequest): return JSONResponse({"ok": False, "error": str(exc)[:500]}, status_code=422) +@router.post("/row/insert") +async def insert_row(body: RowInsertRequest): + if not body.values: + return JSONResponse({"ok": False, "error": "No values provided"}, status_code=400) + try: + return _insert_row(body.engine, body.object, body.values) + except Exception as exc: + return JSONResponse({"ok": False, "error": str(exc)[:500]}, status_code=422) + + @router.post("/benchmark") async def benchmark(): """Parallel analytics: PostgreSQL OLTP vs Trino distributed engine (5M sin() rows).""" diff --git a/ui/src/App.tsx b/ui/src/App.tsx index 3a86619..e105178 100644 --- a/ui/src/App.tsx +++ b/ui/src/App.tsx @@ -18,6 +18,7 @@ import { StorageView } from './components/features/StorageView' import { ChangesView } from './components/features/ChangesView' import { DataFlowView } from './components/features/DataFlowView' import { SearchView } from './components/features/SearchView' +import { DataExplorerView } from './components/features/DataExplorerView' import { DataHubView } from './components/features/DataHubView' import { SshTerminal } from './components/features/SshTerminal' import { TerminalDock } from './components/features/TerminalDock' @@ -141,6 +142,8 @@ export default function App() { ) : cc.mainView === 'search' ? ( + ) : cc.mainView === 'dataexplorer' ? ( + ) : ( )} diff --git a/ui/src/components/features/DataExplorerView.tsx b/ui/src/components/features/DataExplorerView.tsx new file mode 100644 index 0000000..5a011e0 --- /dev/null +++ b/ui/src/components/features/DataExplorerView.tsx @@ -0,0 +1,323 @@ +import { useCallback, useEffect, useMemo, useState } from 'react' +import { + Users, + UserCog, + Package, + ShoppingCart, + DollarSign, + Activity, + Boxes, + RefreshCw, + Search, + Loader2, + TrendingUp, + ExternalLink, +} from 'lucide-react' +import { cn } from '../../lib/utils' + +type Bucket = { key: string; count: number; value?: number } +type Business = { + ok: boolean + generated_at: string + filters: { q: string; region: string } + kpis: { + customers: number + employees: number + products: number + orders_indexed: number + revenue_indexed: number + avg_order: number + telemetry_indexed: number + supply_indexed: number + source_totals: { orders?: number; hr_events?: number; supply_events?: number } + } + orders: { + by_region: Bucket[] + by_channel: Bucket[] + by_status: Bucket[] + top_customers: Bucket[] + top_products: Bucket[] + over_time: Bucket[] + } + hr: { by_department: Bucket[]; by_role: Bucket[]; by_event: Bucket[] } + supply: { by_type: Bucket[]; by_region: Bucket[] } + telemetry: { by_metric: Bucket[] } + regions: string[] +} + +const REGION_COLORS = ['#34d399', '#60a5fa', '#f472b6', '#fbbf24', '#a78bfa', '#22d3ee', '#fb7185'] + +function fmtNum(n?: number | null) { + if (n == null) return '—' + if (Math.abs(n) >= 1_000_000_000) return `${(n / 1_000_000_000).toFixed(2)}B` + if (Math.abs(n) >= 1_000_000) return `${(n / 1_000_000).toFixed(1)}M` + if (Math.abs(n) >= 1_000) return `${(n / 1_000).toFixed(1)}K` + return String(n) +} +function fmtMoney(n?: number | null) { + if (n == null) return '—' + return `€${fmtNum(n)}` +} +function monthLabel(key: string) { + const m = /^(\d{4})-(\d{2})/.exec(key) + if (!m) return key + const months = ['Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun', 'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec'] + return `${months[Number(m[2]) - 1]} '${m[1].slice(2)}` +} + +function Kpi({ icon: Icon, label, value, sub, accent }: { icon: typeof Users; label: string; value: string; sub?: string; accent: string }) { + return ( +
+
+ +
+
+

{label}

+

{value}

+ {sub &&

{sub}

} +
+
+ ) +} + +function Panel({ title, subtitle, children, icon: Icon }: { title: string; subtitle?: string; children: React.ReactNode; icon?: typeof Users }) { + return ( +
+
+ {Icon && } +

{title}

+ {subtitle && {subtitle}} +
+ {children} +
+ ) +} + +function BarsH({ data, valueKind, colorByIndex }: { data: Bucket[]; valueKind?: 'money' | 'num'; colorByIndex?: boolean }) { + const useVal = valueKind != null + const max = Math.max(1, ...data.map((d) => (useVal && d.value != null ? d.value : d.count))) + if (!data.length) return

No data

+ return ( +
+ {data.map((d, i) => { + const metric = useVal && d.value != null ? d.value : d.count + const pct = Math.max(2, (metric / max) * 100) + const color = colorByIndex ? REGION_COLORS[i % REGION_COLORS.length] : '#38bdf8' + const label = useVal && d.value != null ? (valueKind === 'money' ? fmtMoney(d.value) : fmtNum(d.value)) : fmtNum(d.count) + return ( +
+ {d.key ?? '—'} +
+
+
+ {label} +
+ ) + })} +
+ ) +} + +function AreaTrend({ data }: { data: Bucket[] }) { + const pts = data.filter((d) => d.value != null) + if (pts.length < 2) return

Not enough data

+ const w = 560 + const h = 120 + const max = Math.max(...pts.map((p) => p.value || 0)) + const min = Math.min(...pts.map((p) => p.value || 0)) + const range = max - min || 1 + const step = w / (pts.length - 1) + const coords = pts.map((p, i) => [i * step, h - ((((p.value || 0) - min) / range) * (h - 16) + 8)]) + const line = coords.map((c, i) => `${i === 0 ? 'M' : 'L'}${c[0].toFixed(1)},${c[1].toFixed(1)}`).join(' ') + const area = `${line} L${w},${h} L0,${h} Z` + return ( +
+ + + + + + + + + + {coords.map((c, i) => )} + +
+ {monthLabel(pts[0].key)} + {monthLabel(pts[Math.floor(pts.length / 2)].key)} + {monthLabel(pts[pts.length - 1].key)} +
+
+ ) +} + +function Donut({ data }: { data: Bucket[] }) { + const total = data.reduce((s, d) => s + d.count, 0) || 1 + let acc = 0 + const r = 42 + const c = 2 * Math.PI * r + if (!data.length) return

No data

+ return ( +
+ + {data.map((d, i) => { + const frac = d.count / total + const dash = frac * c + const seg = ( + + ) + acc += dash + return seg + })} + +
+ {data.slice(0, 7).map((d, i) => ( +
+ + {d.key ?? '—'} + {((d.count / total) * 100).toFixed(0)}% +
+ ))} +
+
+ ) +} + +export function DataExplorerView() { + const [data, setData] = useState(null) + const [loading, setLoading] = useState(true) + const [region, setRegion] = useState('') + const [qInput, setQInput] = useState('') + const [q, setQ] = useState('') + + const load = useCallback(async () => { + setLoading(true) + try { + const params = new URLSearchParams() + if (q) params.set('q', q) + if (region) params.set('region', region) + const r = await fetch(`/api/search/business?${params.toString()}`) + if (r.ok) setData(await r.json()) + } catch { /* */ } finally { + setLoading(false) + } + }, [q, region]) + + useEffect(() => { load() }, [load]) + + const k = data?.kpis + const regions = useMemo(() => data?.regions || [], [data]) + + return ( +
+ {/* header */} +
+
+

+ + Data Explorer +

+

+ Your actual business data — customers, orders, employees, supply chain & telemetry across all sources +

+
+
+
{ e.preventDefault(); setQ(qInput.trim()) }} + className="flex items-center gap-1 rounded-md border border-border bg-surface px-2 py-1" + > + + setQInput(e.target.value)} + placeholder="Filter by name, id, region…" + className="w-48 bg-transparent text-[11px] text-foreground outline-none placeholder:text-foreground-faint" + /> + + +
+
+ + {/* region filters */} +
+ Region + {['', ...regions].map((rg) => ( + + ))} + {(q || region) && ( + + )} + {data && updated {new Date(data.generated_at).toLocaleTimeString()}} +
+ + {/* KPIs */} +
+ + + + + + + +
+ + {/* Sales */} +

Sales & Customers — PostgreSQL

+
+ + + + + + + + + +
+ + + + + + +
+ + + +
+ + {/* HR */} +

Workforce — MySQL HR

+
+ + + +
+ + {/* Supply + telemetry */} +

Supply Chain (MongoDB) & Telemetry (Cassandra)

+
+ + + +
+
+ ) +} diff --git a/ui/src/components/features/DataSourcesView.tsx b/ui/src/components/features/DataSourcesView.tsx index 09a768f..bf54682 100644 --- a/ui/src/components/features/DataSourcesView.tsx +++ b/ui/src/components/features/DataSourcesView.tsx @@ -1,16 +1,20 @@ import { useCallback, useEffect, useRef, useState } from 'react' import { Activity, + AlertCircle, + CheckCircle2, ChevronRight, Database, FolderTree, Loader2, Network, Play, + Plus, RefreshCw, Server, Table2, TerminalSquare, + X, } from 'lucide-react' import { Badge } from '../ui/Badge' import { DbBrandIcon, dbBrandColor } from '../ui/DbBrandIcon' @@ -94,8 +98,47 @@ export function DataSourcesView({ focusEngine, onPulse }: Props) { const [sampleLoading, setSampleLoading] = useState(false) const [page, setPage] = useState(1) const [pageSize, setPageSize] = useState(100) + const [showInsert, setShowInsert] = useState(false) + const [insertVals, setInsertVals] = useState>({}) + const [inserting, setInserting] = useState(false) + const [insertMsg, setInsertMsg] = useState<{ ok: boolean; text: string } | null>(null) const meta = getSourceMeta(active) + const cdcEngine = active === 'postgres' || active === 'mysql' || active === 'mongodb' + + const submitInsert = useCallback(async () => { + if (!selectedObject) return + setInserting(true) + setInsertMsg(null) + try { + const values: Record = {} + Object.entries(insertVals).forEach(([key, val]) => { if (val !== '') values[key] = val }) + if (Object.keys(values).length === 0) { + setInsertMsg({ ok: false, text: 'Enter at least one value' }) + setInserting(false) + return + } + const r = await fetch('/api/sql/row/insert', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ engine: active, object: selectedObject.fqn, values }), + }) + const j = await r.json() + if (j.ok) { + setInsertMsg({ ok: true, text: cdcEngine ? 'Inserted — Debezium CDC is streaming this change to Kafka → see Live Changes' : 'Inserted successfully' }) + onPulse?.() + loadSample(active, selectedObject, page, pageSize) + setTimeout(() => { setShowInsert(false); setInsertMsg(null) }, 1800) + } else { + setInsertMsg({ ok: false, text: j.error || 'Insert failed' }) + } + } catch { + setInsertMsg({ ok: false, text: 'Insert request failed' }) + } finally { + setInserting(false) + } + // eslint-disable-next-line react-hooks/exhaustive-deps + }, [selectedObject, insertVals, active, cdcEngine, page, pageSize]) useEffect(() => { if (focusEngine) setActive(focusEngine) @@ -308,12 +351,22 @@ export function DataSourcesView({ focusEngine, onPulse }: Props) {
-
+
{selectedObject ? ( <>Data: {selectedObject.fqn} ) : 'Select an object'} + {selectedObject && ( + + )}
}
+ + {showInsert && selectedObject && ( +
setShowInsert(false)}> +
e.stopPropagation()}> +
+
+

+ Insert row +

+

{selectedObject.fqn}

+
+ +
+ + {cdcEngine && ( +
+ This source has CDC — the new row is captured by Debezium and streamed to Kafka in real time. +
+ )} + +
+ {(sample?.columns || []).map((col) => { + const isPk = (sample?.primary_keys || []).includes(col) + return ( +
+ + setInsertVals((v) => ({ ...v, [col]: e.target.value }))} + placeholder="(leave blank to skip)" + className="flex-1 rounded-md border border-border bg-surface px-2 py-1.5 text-[11px] text-foreground outline-none focus:border-docker/50 placeholder:text-foreground-faint" + /> +
+ ) + })} + {!sample?.columns?.length && ( +

No column metadata — open a table in the browser first.

+ )} +
+ + {insertMsg && ( +
+ {insertMsg.ok ? : } + {insertMsg.text} +
+ )} + +
+ + +
+
+
+ )} ) } diff --git a/ui/src/components/layout/SideNav.tsx b/ui/src/components/layout/SideNav.tsx index 2881e15..8006d74 100644 --- a/ui/src/components/layout/SideNav.tsx +++ b/ui/src/components/layout/SideNav.tsx @@ -1,5 +1,5 @@ import { useEffect, useState } from 'react' -import { Database, DatabaseZap, HardDrive, Search, LayoutDashboard, MessageSquare, Server, TerminalSquare, Activity, GitBranch, ExternalLink } from 'lucide-react' +import { Database, DatabaseZap, HardDrive, Search, LayoutDashboard, MessageSquare, Server, TerminalSquare, Activity, GitBranch, ExternalLink, LineChart } from 'lucide-react' import type { GpuStatus, WorkloadData } from '../../types' import type { GpuLiveMetrics } from '../../hooks/useLiveMetrics' import { cn } from '../../lib/utils' @@ -7,7 +7,7 @@ import { viewTabActive, viewTabIdle } from '../../lib/tabActive' import { GpuMatrixPanel } from '../features/GpuMatrixPanel' import { LabHealthPanel } from '../features/LabHealthPanel' -type MainView = 'platform' | 'dataquality' | 'knowledge' | 'storage' | 'hdfs' | 'search' | 'approvals' | 'changes' | 'dataflow' | 'datasources' +type MainView = 'platform' | 'dataquality' | 'knowledge' | 'storage' | 'hdfs' | 'search' | 'approvals' | 'changes' | 'dataflow' | 'datasources' | 'dataexplorer' type Props = { workload: WorkloadData | null @@ -26,6 +26,7 @@ type Props = { const VIEWS: { id: MainView; label: string; icon: typeof LayoutDashboard }[] = [ { id: 'platform', label: 'Data Platform', icon: LayoutDashboard }, { id: 'datasources', label: 'Data Hub', icon: Database }, + { id: 'dataexplorer', label: 'Data Explorer', icon: LineChart }, { id: 'changes', label: 'Live Changes', icon: Activity }, { id: 'dataflow', label: 'Data Flow', icon: GitBranch }, { id: 'dataquality', label: 'Data Quality', icon: DatabaseZap }, diff --git a/ui/src/hooks/useCommandCenter.ts b/ui/src/hooks/useCommandCenter.ts index 48927d1..46f656d 100644 --- a/ui/src/hooks/useCommandCenter.ts +++ b/ui/src/hooks/useCommandCenter.ts @@ -60,7 +60,7 @@ export function useCommandCenter() { const [selectedNode, setSelectedNode] = useState(null) const [nodeDetail, setNodeDetail] = useState(null) const [nodeBusy, setNodeBusy] = useState(false) - const [mainView, setMainView] = useState<'platform' | 'approvals' | 'dataquality' | 'knowledge' | 'storage' | 'hdfs' | 'search' | 'changes' | 'dataflow' | 'datasources'>('platform') + const [mainView, setMainView] = useState<'platform' | 'approvals' | 'dataquality' | 'knowledge' | 'storage' | 'hdfs' | 'search' | 'changes' | 'dataflow' | 'datasources' | 'dataexplorer'>('platform') const [changes, setChanges] = useState([]) const [genPulse, setGenPulse] = useState(false) const genPulseTimer = useRef | null>(null)