Data Flow & Communication Patterns¶

Table of Contents¶

Introduction
Project Structure
Core Components
Architecture Overview
Detailed Component Analysis
Dependency Analysis
Performance Considerations
Troubleshooting Guide
Conclusion
Appendices

Introduction¶

This document explains how data flows and communicates across the application, focusing on: - REST API endpoints and request/response contracts - Real-time updates via Server-Sent Events (SSE) - Internal module interactions between routes, use cases, parsers, matchers, and storage layers - Shared schema definitions and type system - Decision lifecycle from creation to persistence, history tracking, and status synchronization - Caching strategies, message queuing patterns, and event broadcasting for real-time features - Data consistency, transaction management, and conflict resolution approaches - Performance considerations for large datasets and concurrent operations

Project Structure¶

The service layer exposes HTTP APIs and orchestrates business logic. The shared layer defines models and schemas used by both backend and frontend. The database layer persists decisions, history, logs, and users. Cache files provide lightweight state snapshots.

graph TB
subgraph "Service Layer"
A["FastAPI App<br/>apps/service/app.py"]
B["Routes<br/>apps/service/routes/*"]
C["Use Cases<br/>apps/service/core/*_usecases.py"]
D["Parsers & Matchers<br/>apps/service/core/*_parser.py,<br/>apps/service/core/matcher.py,<br/>apps/service/core/confidence.py"]
E["External APIs<br/>apps/service/api/*"]
end
subgraph "Shared Layer"
F["Models<br/>apps/shared/models/*"]
G["Schemas<br/>apps/shared/schemas/*"]
H["DB Config<br/>apps/shared/db.py"]
I["Config<br/>apps/shared/config.py"]
end
subgraph "Persistence"
J["SQLAlchemy DB<br/>src/database/*_db.py"]
K["Cache JSON<br/>cache/*.json"]
end
A --> B
B --> C
C --> D
C --> E
C --> J
C --> K
B --> F
B --> G
A --> H
A --> I

Diagram sources - apps/service/app.py - apps/service/routes/decisions.py - apps/service/core/decision.py - apps/service/core/document_usecases.py - apps/service/core/image_usecases.py - apps/service/api/tmdb.py - apps/shared/models/decision.py - apps/shared/schemas/decision.py - apps/shared/db.py - src/database/decisions_db.py - cache/decisions.json

Section sources - apps/service/main.py - apps/service/app.py - apps/shared/db.py - apps/shared/config.py

Core Components¶

FastAPI application entrypoint and dependency injection setup
Route handlers for authentication, configuration, decisions, history, Jellyfin integration, logs, push notifications, status, tools, and users
Use case modules encapsulating domain workflows (documents, images, general)
Parsers, classifiers, matcher, confidence calculator, organizer, scanner
External integrations (AI, TMDB, TMDB index)
Shared models and Pydantic schemas for consistent contracts
Database access modules for decisions, history, logs, users
Cache files for quick state retrieval

Key responsibilities: - Routes validate requests using schemas and orchestrate use cases - Use cases coordinate parsers, matchers, external APIs, and persistence - Models define entity structures; schemas define API contracts - Database modules implement CRUD and queries - Cache files store lightweight snapshots for performance

Architecture Overview¶

High-level flow: - Client sends HTTP requests to FastAPI routes - Routes validate payloads with Pydantic schemas and call use cases - Use cases execute parsing, matching, confidence scoring, and organization - External services (AI, TMDB) are invoked as needed - Results are persisted via database modules and optionally cached - Status and events are broadcast via SSE to clients

sequenceDiagram
participant Client as "Client"
participant API as "FastAPI App"
participant Route as "Route Handler"
participant UC as "Use Case"
participant Ext as "External API"
participant DB as "Database Module"
participant Cache as "Cache File"
Client->>API : "HTTP Request"
API->>Route : "Dispatch to route"
Route->>Route : "Validate with Schema"
Route->>UC : "Invoke workflow"
UC->>Ext : "Call AI/TMDB if needed"
Ext-->>UC : "Response"
UC->>DB : "Persist decision/history"
DB-->>UC : "Success/Failure"
UC->>Cache : "Update snapshot"
Cache-->>UC : "OK"
UC-->>Route : "Result DTO"
Route-->>Client : "HTTP Response"

Diagram sources - apps/service/app.py - apps/service/routes/decisions.py - apps/service/core/document_usecases.py - apps/service/api/tmdb.py - src/database/decisions_db.py - cache/decisions.json

Detailed Component Analysis¶

REST API Communications¶

Authentication: login/logout and user management endpoints
Configuration: read/update settings
Decisions: create, list, update, delete decisions
History: query historical records
Jellyfin: sync and manage media server state
Logs: retrieve operational logs
Push: subscribe/unsubscribe to notifications
Status: get current system status
Tools: utility endpoints

Request validation is performed using Pydantic schemas from the shared layer. Responses are typed and consistent.

flowchart TD
Start(["Incoming Request"]) --> Validate["Validate Payload with Schema"]
Validate --> Valid{"Valid?"}
Valid --> |No| Err["Return Validation Error"]
Valid --> |Yes| CallUC["Call Use Case"]
CallUC --> Persist["Persist via DB Module"]
Persist --> UpdateCache["Update Cache Snapshot"]
UpdateCache --> Respond["Return Typed Response"]
Err --> End(["Exit"])
Respond --> End

Diagram sources - apps/service/routes/decisions.py - apps/service/routes/history.py - apps/service/routes/status.py - apps/service/routes/push.py - apps/service/routes/jellyfin.py - apps/service/routes/logs.py - apps/service/routes/config.py - apps/service/routes/users.py - apps/service/routes/auth.py - apps/service/routes/tools.py - apps/shared/schemas/decision.py - apps/shared/schemas/history.py - apps/shared/schemas/config.py - apps/shared/schemas/user.py - apps/shared/schemas/auth.py

Section sources - apps/service/routes/decisions.py - apps/service/routes/history.py - apps/service/routes/status.py - apps/service/routes/push.py - apps/service/routes/jellyfin.py - apps/service/routes/logs.py - apps/service/routes/config.py - apps/service/routes/users.py - apps/service/routes/auth.py - apps/service/routes/tools.py - apps/shared/schemas/decision.py - apps/shared/schemas/history.py - apps/shared/schemas/config.py - apps/shared/schemas/user.py - apps/shared/schemas/auth.py

Real-Time Updates via Server-Sent Events (SSE)¶

SSE endpoints stream status changes and event notifications to connected clients
Clients maintain persistent connections to receive live updates without polling
Events include scan progress, decision outcomes, and system status transitions

sequenceDiagram
participant Client as "Browser/Client"
participant SSE as "SSE Endpoint"
participant Bus as "Event Broadcast"
participant Status as "Status Store"
Client->>SSE : "Connect /events"
SSE->>Bus : "Subscribe to events"
Status-->>Bus : "Emit status change"
Bus-->>SSE : "Forward event"
SSE-->>Client : "event : status\nData : {...}"

Diagram sources - apps/service/routes/status.py - apps/service/routes/push.py - cache/status.json

Section sources - apps/service/routes/status.py - apps/service/routes/push.py - cache/status.json

Internal Module Interactions¶

Use cases coordinate parsers, classifiers, matchers, and organizers
Confidence calculation influences decision outcomes
External APIs enrich metadata (TMDB, AI)
Database modules persist structured entities
Cache files provide fast reads for dashboards and UI

classDiagram
class DecisionUseCases {
+create_decision(data)
+update_decision(id, data)
+list_decisions(filters)
}
class DocumentUseCases {
+parse_document(path)
+classify_document(doc)
}
class ImageUseCases {
+parse_image(path)
+classify_image(img)
}
class Matcher {
+match(decision, candidates)
}
class Confidence {
+score(match_result)
}
class Organizer {
+organize(decision)
}
class Parser {
+extract_metadata(file)
}
class DocumentParser {
+parse_text(text)
}
class ImageParser {
+parse_binary(data)
}
class ImageClassifier {
+predict(image)
}
class Scanner {
+scan_directory(path)
}
class TmdbApi {
+search(query)
}
class AiApi {
+analyze(content)
}
DecisionUseCases --> Matcher : "uses"
DecisionUseCases --> Confidence : "uses"
DecisionUseCases --> Organizer : "uses"
DocumentUseCases --> DocumentParser : "uses"
ImageUseCases --> ImageParser : "uses"
ImageUseCases --> ImageClassifier : "uses"
DecisionUseCases --> Parser : "uses"
DecisionUseCases --> Scanner : "uses"
DecisionUseCases --> TmdbApi : "calls"
DecisionUseCases --> AiApi : "calls"

Diagram sources - apps/service/core/decision.py - apps/service/core/document_usecases.py - apps/service/core/image_usecases.py - apps/service/core/usecases.py - apps/service/core/matcher.py - apps/service/core/confidence.py - apps/service/core/organizer.py - apps/service/core/parser.py - apps/service/core/document_parser.py - apps/service/core/image_parser.py - apps/service/core/image_classifier.py - apps/service/core/scan.py - apps/service/api/tmdb.py - apps/service/api/ai.py

Section sources - apps/service/core/decision.py - apps/service/core/document_usecases.py - apps/service/core/image_usecases.py - apps/service/core/usecases.py - apps/service/core/matcher.py - apps/service/core/confidence.py - apps/service/core/organizer.py - apps/service/core/parser.py - apps/service/core/document_parser.py - apps/service/core/image_parser.py - apps/service/core/image_classifier.py - apps/service/core/scan.py - apps/service/api/tmdb.py - apps/service/api/ai.py

Shared Schema Definitions and Type System¶

Models define core entities (Decision, History, ActionLog, User, PushSubscription, Status)
Schemas define API contracts (Auth, Config, Decision, History, User)
Types module centralizes shared enums and constants
Consistent typing ensures reliable serialization/deserialization across components

erDiagram
DECISION {
uuid id PK
string title
string type
text description
float confidence
enum status
timestamp created_at
timestamp updated_at
}
HISTORY {
uuid id PK
uuid decision_id FK
string action
text details
timestamp occurred_at
}
ACTION_LOG {
uuid id PK
string level
string message
timestamp logged_at
}
USER {
uuid id PK
string username UK
string password_hash
boolean active
timestamp created_at
}
PUSH_SUBSCRIPTION {
uuid id PK
string endpoint
string keys_p256dh
string keys_auth
timestamp subscribed_at
}
STATUS {
string key PK
json value
timestamp updated_at
}
DECISION ||--o{ HISTORY : "has many"

Diagram sources - apps/shared/models/decision.py - apps/shared/models/history.py - apps/shared/models/action_log.py - apps/shared/models/user.py - apps/shared/models/push_subscription.py - apps/shared/models/status.py - apps/shared/schemas/decision.py - apps/shared/schemas/history.py - apps/shared/schemas/config.py - apps/shared/schemas/user.py - apps/shared/schemas/auth.py - apps/service/core/types.py

Section sources - apps/shared/models/decision.py - apps/shared/models/history.py - apps/shared/models/action_log.py - apps/shared/models/user.py - apps/shared/models/push_subscription.py - apps/shared/models/status.py - apps/shared/schemas/decision.py - apps/shared/schemas/history.py - apps/shared/schemas/config.py - apps/shared/schemas/user.py - apps/shared/schemas/auth.py - apps/service/core/types.py

Decision Lifecycle: Creation to Persistence and History Tracking¶

End-to-end flow: - Client submits a decision via REST - Route validates input using schemas - Use case orchestrates parsing, classification, matching, and confidence scoring - External APIs enrich metadata when applicable - Decision is persisted via database module - History record is created for auditability - Status cache is updated and SSE broadcasts the new state

sequenceDiagram
participant Client as "Client"
participant Route as "Decisions Route"
participant UC as "Decision Use Case"
participant Parser as "Parser/Document/Image Parser"
participant Classifier as "Image Classifier"
participant Matcher as "Matcher"
participant Confidence as "Confidence Calculator"
participant DB as "Decisions DB"
participant HistDB as "History DB"
participant Cache as "Cache"
participant SSE as "SSE"
Client->>Route : "POST /decisions"
Route->>Route : "Validate with Decision Schema"
Route->>UC : "Create decision"
UC->>Parser : "Extract metadata"
Parser-->>UC : "Metadata"
UC->>Classifier : "Classify image if needed"
Classifier-->>UC : "Labels"
UC->>Matcher : "Match against candidates"
Matcher-->>UC : "Matches"
UC->>Confidence : "Compute score"
Confidence-->>UC : "Score"
UC->>DB : "Persist decision"
DB-->>UC : "ID"
UC->>HistDB : "Record history entry"
HistDB-->>UC : "OK"
UC->>Cache : "Update status/decisions snapshot"
Cache-->>UC : "OK"
UC->>SSE : "Broadcast event"
SSE-->>Client : "Real-time update"
Route-->>Client : "Created decision"

Diagram sources - apps/service/routes/decisions.py - apps/service/core/decision.py - apps/service/core/document_usecases.py - apps/service/core/image_usecases.py - apps/service/core/parser.py - apps/service/core/document_parser.py - apps/service/core/image_parser.py - apps/service/core/image_classifier.py - apps/service/core/matcher.py - apps/service/core/confidence.py - src/database/decisions_db.py - src/database/history_db.py - cache/decisions.json - cache/status.json

Section sources - apps/service/routes/decisions.py - apps/service/core/decision.py - apps/service/core/document_usecases.py - apps/service/core/image_usecases.py - apps/service/core/parser.py - apps/service/core/document_parser.py - apps/service/core/image_parser.py - apps/service/core/image_classifier.py - apps/service/core/matcher.py - apps/service/core/confidence.py - src/database/decisions_db.py - src/database/history_db.py - cache/decisions.json - cache/status.json

Status Synchronization and Event Broadcasting¶

Status module provides endpoints to read current system state
SSE endpoint streams status changes to clients
Push subscriptions allow out-of-band notifications
Cache stores latest status for fast reads

flowchart TD
Read["GET /status"] --> CacheRead["Read cache/status.json"]
CacheRead --> Return["Return status object"]
Update["Background process updates status"] --> CacheWrite["Write cache/status.json"]
CacheWrite --> SSE["Emit SSE event"]
SSE --> Clients["Connected clients receive update"]

Diagram sources - apps/service/routes/status.py - apps/service/routes/push.py - cache/status.json

Section sources - apps/service/routes/status.py - apps/service/routes/push.py - cache/status.json

Caching Strategies¶

Lightweight JSON cache files store frequently accessed data (decisions, history, status, Jellyfin state)
Cache updates occur after persistence to ensure eventual consistency
Reads bypass heavy computations by serving cached snapshots

flowchart TD
Write["Persist to DB"] --> UpdateCache["Update cache file"]
UpdateCache --> Serve["Serve from cache on subsequent reads"]

Diagram sources - cache/decisions.json - cache/history.json - cache/status.json - cache/jellyfin_state.json

Section sources - cache/decisions.json - cache/history.json - cache/status.json - cache/jellyfin_state.json

Message Queuing Patterns and Event Broadcasting¶

SSE acts as an event bus for real-time updates
Push subscription endpoints manage client endpoints for notifications
Background processes emit events that are forwarded to connected clients

sequenceDiagram
participant Worker as "Background Worker"
participant Bus as "Event Bus"
participant SSE as "SSE Endpoint"
participant Client as "Client"
Worker->>Bus : "Publish event"
Bus->>SSE : "Deliver to subscribers"
SSE->>Client : "Stream event"

Diagram sources - apps/service/routes/status.py - apps/service/routes/push.py

Section sources - apps/service/routes/status.py - apps/service/routes/push.py

Data Consistency, Transaction Management, and Conflict Resolution¶

Database modules encapsulate transactions around write operations
History records provide an immutable audit trail for decisions
Cache writes follow successful DB commits to maintain consistency
Conflict resolution relies on unique identifiers and versioned updates where applicable

flowchart TD
BeginTx["Begin Transaction"] --> WriteDB["Write Decision"]
WriteDB --> WriteHistory["Write History Entry"]
WriteHistory --> Commit["Commit Transaction"]
Commit --> UpdateCache["Update Cache"]
UpdateCache --> Success["Return Success"]
Commit --> Rollback["Rollback on Error"]

Diagram sources - src/database/decisions_db.py - src/database/history_db.py - cache/decisions.json

Section sources - src/database/decisions_db.py - src/database/history_db.py - cache/decisions.json

Dependency Analysis¶

Component coupling and cohesion: - Routes depend on schemas and use cases - Use cases depend on parsers, classifiers, matchers, confidence calculators, and external APIs - Database modules depend on shared models and configuration - Cache files are consumed by routes and background processes

graph LR
Routes["Routes"] --> Schemas["Schemas"]
Routes --> UseCases["Use Cases"]
UseCases --> Parsers["Parsers"]
UseCases --> Matchers["Matchers"]
UseCases --> Confidence["Confidence"]
UseCases --> External["External APIs"]
UseCases --> DB["Database Modules"]
DB --> Models["Models"]
DB --> Config["Config"]
Routes --> Cache["Cache Files"]

Diagram sources - apps/service/routes/decisions.py - apps/service/core/decision.py - apps/service/core/matcher.py - apps/service/core/confidence.py - apps/service/api/tmdb.py - apps/shared/schemas/decision.py - apps/shared/models/decision.py - src/database/decisions_db.py - cache/decisions.json

Section sources - apps/service/routes/decisions.py - apps/service/core/decision.py - apps/service/core/matcher.py - apps/service/core/confidence.py - apps/service/api/tmdb.py - apps/shared/schemas/decision.py - apps/shared/models/decision.py - src/database/decisions_db.py - cache/decisions.json

Performance Considerations¶

Prefer cache reads for dashboard views to reduce DB load
Batch operations where possible to minimize transaction overhead
Stream large responses via SSE instead of polling
Limit external API calls with caching or rate limiting
Index frequently queried fields in the database
Avoid synchronous blocking in route handlers; offload heavy work to background tasks

[No sources needed since this section provides general guidance]

Troubleshooting Guide¶

Common issues and diagnostics: - Validation errors: check schema definitions and payload structure - SSE disconnections: verify endpoint availability and network stability - Cache inconsistencies: ensure cache updates follow successful DB commits - External API failures: inspect error handling in API modules and retry policies - Database errors: review transaction boundaries and rollback paths

Section sources - apps/service/routes/decisions.py - apps/service/routes/status.py - apps/service/api/tmdb.py - src/database/decisions_db.py - cache/decisions.json

Conclusion¶

The application employs a layered architecture with clear separation of concerns: - REST APIs enforce contracts through shared schemas - Use cases orchestrate complex workflows involving parsers, matchers, and external services - Database modules handle persistence with transactional integrity - SSE enables real-time updates while cache files improve read performance - History tracking ensures auditability and supports conflict resolution

This design balances scalability, reliability, and responsiveness, making it suitable for managing large datasets and concurrent operations.

[No sources needed since this section summarizes without analyzing specific files]

Appendices¶

Configuration and environment variables are managed via shared config modules
Alembic migrations define schema evolution
Documentation artifacts describe step-by-step development milestones

[No sources needed since this section provides general guidance]