Sơ đồ kiến trúc pipeline CPG streaming

1Mục tiêu¶

Trang này tổng hợp kiến trúc của Lab 04 ở mức hệ thống. Mục tiêu là theo dõi một file Python từ lúc được discovery trong repository nguồn, qua Parser Service tăng dần, tới các event Kafka và hai nhánh downstream: Neo4j cho graph topology và MongoDB cho metadata.

Kiến trúc nhấn mạnh ba thuộc tính chính: đầu vào có thể tái hiện bằng commit SHA, event contract ổn định bằng stable identifiers, và replay-safe behavior bằng state, upsert, checkpoint, tombstones và constraints.

2Tổng quan hệ thống¶

Pipeline bắt đầu bằng repository discovery từ source root. Discovery tạo raw Python inventory, áp dụng file filters để xác định eligible parser inputs, rồi lưu kết quả vào manifest. Parser Service đọc manifest, xử lý từng file độc lập và publish các event đã validate schema vào Kafka.

Từ Kafka, graph events đi trực tiếp vào Neo4j bằng Kafka Connect Sink. Metadata events đi qua Spark Structured Streaming và được upsert vào MongoDB. Hai nhánh downstream dùng cùng file_id và generation metadata nhưng hoạt động độc lập sau khi event đã được Kafka acknowledgement.

3Thành phần chính¶

Thành phần	Vai trò
Source repository	Snapshot mã nguồn được clone shallow và gắn với commit SHA cố định.
Repository discovery	Enumerate Python files từ repository root và ghi raw inventory.
File filters	Xác định record hợp lệ cho parser và ghi reason cho record bị loại.
Discovery manifest	Source of truth nối Task 1 với Parser Service ở Task 2.
Parser Service	Parse từng file, sinh AST/CFG/DFG/call graph, metadata và parser errors.
SQLite state store	Lưu content hash và graph state để hỗ trợ incremental skip/replay.
Kafka topics	Phân tách node, edge, metadata và parser error streams.
Kafka Connect Neo4j sinks	Ghi graph vào Neo4j bằng Cypher `MERGE`/`DELETE`.
Spark Structured Streaming	Consume metadata stream, checkpoint offsets và upsert MongoDB.

4Kiến trúc pipeline tổng thể¶

5Discovery và Parser Service¶

Discovery tạo raw inventory trước khi áp dụng filter. Cách này giúp báo cáo được cả quy mô repository lẫn phạm vi parser thực tế. Manifest lưu path tương đối, commit SHA, file size, content hash, trạng thái included và exclusion_reason.

Parser Service đọc các eligible records từ manifest và xử lý từng file. Với file chưa đổi, service dùng content hash trong SQLite state để skip. Với file mới hoặc đã thay đổi, service parse bằng Python ast, sinh graph entities, validate event schema, ghi ra JSONL dry-run hoặc publish Kafka, rồi chỉ commit state sau khi writer/Kafka acknowledgement thành công.

6Kafka event streams¶

Topic	Producer	Nội dung	Kafka key
`cpg.nodes`	Parser Service	Node upsert/delete events	`file_id`
`cpg.edges`	Parser Service	Edge upsert/delete events	`file_id`
`source.metadata`	Parser Service	File metadata events	`file_id`
`parser.errors`	Parser Service	Lỗi phân tích source hoặc parser business errors	`file_id`
`connector.errors`	Kafka Connect	Dead-letter records từ downstream ingestion	connector key

Event envelope chứa schema_version, event_time, stable event identifiers, file_id, content_hash, parser metadata và payload theo từng loại event. Kafka key bằng file_id giúp routing nhất quán trong từng topic; ordering được giữ trong partition của topic đó.

7Neo4j graph ingestion¶

Node và edge events đi trực tiếp từ Kafka sang Neo4j qua Kafka Connect, không đi qua Spark. Connector dùng Cypher MERGE để upsert stable node/edge IDs, uniqueness constraints để ngăn duplicate, placeholder nodes để xử lý edge đến trước node, và tombstones để chặn stale replay trong cùng generation.

connector.errors là dead-letter topic của Kafka Connect. Topic này khác với parser.errors, vốn do Parser Service publish khi không thể phân tích một file nguồn.

8Spark/MongoDB metadata ingestion¶

Metadata stream source.metadata được consume bởi Spark Structured Streaming. Spark dùng checkpoint để ghi nhớ offsets đã xử lý và upsert document vào MongoDB theo file_id. Nhánh này phục vụ truy vấn metadata file, commit, content hash, parser version và trạng thái parse mà không trộn lẫn với graph topology trong Neo4j.

9Incremental replay flow¶

10Các cơ chế nổi bật¶

Cơ chế	Vai trò
Commit SHA	Truy vết source snapshot
Content hash	Phát hiện file thay đổi
Stable IDs	Giữ định danh node/edge qua replay
SQLite state	Lưu trạng thái parser tăng dần
Kafka key = `file_id`	Routing nhất quán trong từng topic
Neo4j `MERGE`	Upsert graph không nhân đôi
Placeholder nodes	Xử lý edge đến trước node
Tombstones	Chặn stale replay cùng generation
Spark checkpoint	Khôi phục offsets đã xử lý
MongoDB upsert	Cập nhật metadata theo `file_id`

11Ranh giới các layer¶

Layer	Trách nhiệm	Quy tắc phụ thuộc
`domain/`	Model, enum, event contract và lỗi nghiệp vụ.	Không phụ thuộc layer khác.
`parsing/`	AST, CFG, DFG, call graph, stable IDs và diff.	Chỉ phụ thuộc `domain`.
`application/`	Use case service và port interface.	Giao tiếp qua ports, không khởi tạo adapter cụ thể.
`infrastructure/`	Kafka, JSONL writer, SQLite, config và filesystem adapters.	Implement ports từ application.
`cli/`	Composition root, load config và inject adapters.	Được phép nối các layer khi chạy command.
`spark_jobs/`	Spark Structured Streaming job cho MongoDB.	Độc lập với parser core.

Kafka, SQLite, Neo4j và MongoDB không tham gia cùng một distributed transaction. Hệ thống sử dụng stable IDs, upsert, checkpoint và tombstones để đạt replay-safe behavior cho các kịch bản đã kiểm chứng.

12Kết quả kiến trúc¶

Kiến trúc hiện tại tách rõ discovery, parsing, event distribution và downstream ingestion. Task 1 tạo manifest có thể truy vết; Task 2 tạo event contract ổn định; Task 3 xác minh Kafka topic routing; Task 4 xác minh graph ingestion trực tiếp vào Neo4j. Nhánh Spark/MongoDB dùng metadata stream và checkpoint để phục vụ Task 5.

Thiết kế này giúp mỗi task có evidence riêng nhưng vẫn nối với nhau bằng các contract cụ thể: manifest paths, stable IDs, Kafka topics, schema version, generation metadata và downstream upsert semantics.

13Nhận xét¶

Điểm mạnh của pipeline là mỗi biên hệ thống đều có cơ chế tái hiện hoặc kiểm chứng: commit SHA ở source snapshot, manifest ở discovery, SQLite state ở parser, Kafka offsets ở event streams, Neo4j constraints ở graph ingestion và Spark checkpoint ở metadata ingestion.

Báo cáo không tuyên bố transaction phân tán toàn trình. Thay vào đó, hệ thống chứng minh replay-safe behavior trong các kịch bản đã kiểm chứng bằng stable identifiers, idempotent writes và state/checkpoint rõ ràng.