CodeVetter · Repo Unpack · Behavior traces
langgenius/dify
Three flows traced end-to-end: (1) API server startup, (2) end-user streaming chat message across the request→Celery→SSE boundary, and (3) the `flask db` migration path plus Celery worker boot. The dominant pattern is 'authenticate at the edge, rehydrate trusted state inside the worker, stream events back through Redis'.
- STARTUP (gunicorn/prod): `api/docker/entrypoint.sh` → gunicorn gevent workers (`gunicorn.conf.py`) → import `app.py` → `app_factory.create_app()` → `create_flask_app_with_configs()` builds `DifyApp` + pushes `dify_config.model_dump()` into `app.config` + registers `before_request` (request-context init, enterprise license gate) and `after_request` (OTel trace headers) → `initialize_extensions()` runs the ordered ext list → `socketio.WSGIApp(sio, app)` → served on :5001.
- STARTUP (bare `python -m app`): monkey.patch_all() → grpc_gevent.init_gevent() → psycogreen.patch_psycopg() → create_app() → `log_startup_banner` → `pywsgi.WSGIServer((0.0.0.0,5001), socketio_app, WebSocketHandler).serve_forever()`.
- MIGRATION PATH: `flask db upgrade` → `is_db_command()` detects `sys.argv` → `create_migrations_app()` → `create_flask_app_with_configs()` → only `ext_database`, `ext_migrate`, `ext_commands` initialized → Flask-Migrate/Alembic runs `api/migrations/env.py` against the 219-version history.
- WORKER BOOT: celery invokes `celery_entrypoint.py` → gevent/grpc/psycopg patch → `from app import app, celery` (full create_app ran) → `ext_celery.init_app` built the Celery app earlier with `FlaskTask` (app_context + `init_request_context` per task) → `setup_workflow_warm_shutdown_handler()` → beat schedule assembled conditionally from ~20 `ENABLE_*`/edition flags (cleanup crons, tidb serverless, human-input timeouts, plugin upgrade checks, telemetry heartbeats).
- CHAT MESSAGE (request side): POST `/api/chat-messages` → `validate_jwt_token` → `decode_jwt_token` → `extract_webapp_passport` header → `PassportService().verify` → DB loads `App`/`Site`/`EndUser` → webapp-auth mode checks (`_validate_webapp_token`, `_validate_user_accessibility` incl. SSO `granted_at` staleness) → `ChatApi.post` → mode gate → payload validate → eager `ConversationService.get_conversation` → `AppGenerateService.generate`.
- CHAT MESSAGE (guardrails+dispatch): `generate` → `_run_with_guardrails` → `QuotaService.reserve` (CLOUD only) → `RateLimit.enter(request_id)` → `_dispatch_generate` → for `ADVANCED_CHAT`/`WORKFLOW` streaming: `_get_workflow` (published vs draft by invoke_from) → `_ensure_workflow_service_mode_available` (reject trigger nodes for manual invokes) → `rate_limit_context` → `AppExecutionParams.new(...).model_dump_json()` → `_build_streaming_task_on_subscribe` wraps `workflow_based_app_execution_task.delay` → `retrieve_events(app_mode, workflow_run_id, on_subscribe)` → `rate_limit.generate(convert_to_event_stream(...))` → `compact_generate_response` SSE.
- CHAT MESSAGE (worker side): `workflow_based_app_execution_task` deserializes `AppExecutionParams` → opens fresh `Session` → reloads `App`/`Workflow`/user (`set_login_user` for flask-login context) → `WorkflowAppGenerator().generate` → `WorkflowAppConfigManager` + `WorkflowAppRunner` → `WorkflowEntry.__init__` (call-depth check, `InMemoryChannel` default, `capture_current_context`) → `GraphEngine.run()` → `iter_dify_graph_engine_events` filters → queue manager publishes to Redis → client SSE relay → finish persists `WorkflowRun`/node executions; `PauseStatePersistenceLayer` handles human-input pauses.
- DEBUG SINGLE-NODE: `AppGenerateService.generate_single_iteration`/`generate_single_loop` bypass the Celery hop entirely — they resolve the draft workflow via `_get_workflow(..., InvokeFrom.DEBUGGER)` and call `single_iteration_generate`/`single_loop_generate` on the advanced-chat/workflow generators synchronously in the request process.
- INNER API AUTH: plugin-daemon/enterprise calls hit `/inner/api/*` → `inner_api_only` checks `dify_config.INNER_API` (404 if disabled) then constant-compares `X-Inner-Api-Key` against `INNER_API_KEY`; `enterprise_inner_api_user_auth` additionally accepts `Authorization: <user_id>:<hmac>` and injects an `EndUser`.
- SERVICE API AUTH: `/v1/*` resources use flask-restx wrappers that extract the `Authorization` bearer `ApiToken`, cache via `ApiTokenCache`/`fetch_token_with_single_flight`, record token usage, and resolve the `EndUser`-equivalent caller from `user` query/json/form params per `FetchUserArg` policy. (inferred)
- STOP FLOW: `POST /chat-messages/<task_id>/stop` → `AppTaskService.stop_task(task_id, invoke_from, user_id, app_mode)` writes a Redis cancellation key → running task pipeline notices and raises `GenerateTaskStoppedError` → worker cleanup calls `clear_app_task_cancellation_signals`. (inferred)
- EXTENSION INIT ORDER is load-bearing: `ext_key_provider` runs after `ext_storage` because `RSAKeyProvider` reads private keys from storage; `ext_logstore` runs after storage and before celery; `ext_import_modules` runs early so controller modules' route decorators evaluate before blueprint registration.
- EDITION GATING: `DeploymentEdition` (COMMUNITY/CLOUD/ENTERPRISE) is checked at multiple layers — license gate in `before_request`, quota reserve in `_run_with_guardrails`, telemetry beat tasks, and `EnterpriseService`/`SystemFeatureService` lookups for webapp auth and billing features (`FeatureService.get_features` fulfills plan params from env, workspace info, or billing API).