feat: module-driven onboarding system + simplified 3-step signup

Add OnboardingProviderProtocol so modules declare their own post-signup
onboarding steps. The core OnboardingAggregator discovers enabled
providers and exposes a dashboard API (GET /dashboard/onboarding).
A session-scoped banner on the store dashboard shows a checklist that
guides merchants through setup without blocking signup.

Signup is simplified from 4 steps to 3 (Plan → Account → Payment):
store creation is merged into account creation, store language is
captured from the user's browsing language, and platform-specific
template branching is removed.

Includes 47 unit and integration tests covering all new providers,
the aggregator, the API endpoint, and the signup service changes.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-02-28 23:39:42 +01:00
parent f8a2394da5
commit ef9ea29643
26 changed files with 2055 additions and 699 deletions

View File

@@ -0,0 +1,157 @@
# app/modules/core/services/onboarding_aggregator.py
"""
Onboarding aggregator service for collecting onboarding steps from all modules.
This service lives in core because the dashboard onboarding banner is core
functionality. It discovers OnboardingProviders from all enabled modules and
provides a unified interface for the dashboard checklist.
Usage:
from app.modules.core.services.onboarding_aggregator import onboarding_aggregator
summary = onboarding_aggregator.get_onboarding_summary(
db=db, store_id=123, platform_id=1, store_code="my-store"
)
# Returns: {steps, total_steps, completed_steps, progress_percentage, all_completed}
"""
import logging
from typing import TYPE_CHECKING, Any
from sqlalchemy.orm import Session
from app.modules.contracts.onboarding import (
OnboardingProviderProtocol,
OnboardingStepStatus,
)
if TYPE_CHECKING:
from app.modules.base import ModuleDefinition
logger = logging.getLogger(__name__)
class OnboardingAggregatorService:
"""
Aggregates onboarding steps from all module providers.
Discovers OnboardingProviders from enabled modules and provides
a unified interface for the dashboard onboarding banner.
"""
def _get_enabled_providers(
self, db: Session, platform_id: int
) -> list[tuple["ModuleDefinition", OnboardingProviderProtocol]]:
"""
Get onboarding providers from enabled modules.
Returns:
List of (module, provider) tuples for enabled modules with providers
"""
from app.modules.registry import MODULES
from app.modules.service import module_service
providers: list[tuple[ModuleDefinition, OnboardingProviderProtocol]] = []
for module in MODULES.values():
if not module.has_onboarding_provider():
continue
# Core modules are always enabled, check others
if not module.is_core:
try:
if not module_service.is_module_enabled(db, platform_id, module.code):
continue
except Exception as e:
logger.warning(
f"Failed to check if module {module.code} is enabled: {e}"
)
continue
try:
provider = module.get_onboarding_provider_instance()
if provider is not None:
providers.append((module, provider))
except Exception as e:
logger.warning(
f"Failed to get onboarding provider for module {module.code}: {e}"
)
return providers
def get_onboarding_steps(
self, db: Session, store_id: int, platform_id: int
) -> list[OnboardingStepStatus]:
"""
Get all onboarding steps with completion status, sorted by order.
Returns:
Sorted list of OnboardingStepStatus objects
"""
providers = self._get_enabled_providers(db, platform_id)
steps: list[OnboardingStepStatus] = []
for module, provider in providers:
try:
step_defs = provider.get_onboarding_steps()
for step_def in step_defs:
try:
completed = provider.is_step_completed(db, store_id, step_def.key)
except Exception as e:
logger.warning(
f"Failed to check step {step_def.key} from {module.code}: {e}"
)
completed = False
steps.append(OnboardingStepStatus(step=step_def, completed=completed))
except Exception as e:
logger.warning(
f"Failed to get onboarding steps from module {module.code}: {e}"
)
steps.sort(key=lambda s: s.step.order)
return steps
def get_onboarding_summary(
self,
db: Session,
store_id: int,
platform_id: int,
store_code: str = "",
) -> dict[str, Any]:
"""
Get onboarding summary for the dashboard API.
Returns:
Dict with: steps[], total_steps, completed_steps, progress_percentage, all_completed
"""
step_statuses = self.get_onboarding_steps(db, store_id, platform_id)
total = len(step_statuses)
completed = sum(1 for s in step_statuses if s.completed)
steps_data = []
for status in step_statuses:
route = status.step.route_template.replace("{store_code}", store_code)
steps_data.append({
"key": status.step.key,
"title_key": status.step.title_key,
"description_key": status.step.description_key,
"icon": status.step.icon,
"route": route,
"completed": status.completed,
"category": status.step.category,
})
return {
"steps": steps_data,
"total_steps": total,
"completed_steps": completed,
"progress_percentage": round((completed / total * 100) if total > 0 else 0),
"all_completed": completed == total and total > 0,
}
# Singleton instance
onboarding_aggregator = OnboardingAggregatorService()
__all__ = ["OnboardingAggregatorService", "onboarding_aggregator"]