REDCap Data Entry Trigger (DET) Webhook Implementation Plan¶
Overview¶
This document outlines the implementation plan for integrating REDCap-managed studies with Metricis using REDCap's Data Entry Trigger (DET) webhook mechanism. In this model:
- REDCap is the source of truth for participant enrollment
- Metricis handles assessment delivery, notifications, and data collection
- REDCap receives completed assessment data via API push
Architecture¶
┌─────────────────────────────────────────────────────────────────────────────┐
│ REDCap-Managed Study Flow │
└─────────────────────────────────────────────────────────────────────────────┘
┌──────────────┐ DET Webhook ┌──────────────┐
│ REDCap │ ─────────────────────▶│ Metricis │
│ (Project) │ │ Webhook │
│ │ │ Endpoint │
│ • New record │ │ │
│ • Event data │ └──────┬───────┘
└──────────────┘ │
▼
┌──────────────────┐
│ Participant │
│ Sync Service │
│ │
│ • Fetch full │
│ record from │
│ REDCap API │
│ • Create/update │
│ participant │
└────────┬─────────┘
│
▼
┌──────────────────┐
│ Schedule │
│ Service │
│ │
│ • Map events to │
│ batteries │
│ • Calculate │
│ visit dates │
│ • Create visits │
└────────┬─────────┘
│
▼
┌──────────────────┐
│ Notification │
│ Worker │
│ │
│ • Send reminders │
│ • Include │
│ battery links │
└────────┬─────────┘
│
▼
┌──────────────────┐
Participant ◀───── Link ───────────│ Metricis App │
│ (Web/Mobile) │
│ │
│ • Complete │
│ battery │
└────────┬─────────┘
│
▼
┌──────────────┐ API Import ┌──────────────────┐
│ REDCap │ ◀───────────────────│ REDCap Sync │
│ (Project) │ │ Service │
│ │ │ │
│ • Form data │ │ • Map to fields │
│ • Complete │ │ • Push data │
│ status │ │ • Mark complete │
└──────────────┘ └──────────────────┘
Phase 1: DET Webhook Endpoint¶
1.1 Webhook Router¶
Create a new router for REDCap webhooks.
File: server/app/routers/webhooks.py
from fastapi import APIRouter, Request, HTTPException, BackgroundTasks, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.database import get_db
from app.services.redcap_sync import REDCapSyncService
import structlog
router = APIRouter(prefix="/api/webhooks", tags=["webhooks"])
logger = structlog.get_logger()
@router.post("/redcap/det")
async def redcap_data_entry_trigger(
request: Request,
background_tasks: BackgroundTasks,
db: AsyncSession = Depends(get_db)
):
"""
REDCap Data Entry Trigger (DET) webhook endpoint.
REDCap sends POST with form data:
- project_id: REDCap project ID
- record: Record ID (participant identifier)
- instrument: Form/instrument name that was saved
- redcap_event_name: Event name (for longitudinal projects)
- redcap_data_access_group: DAG if applicable
- [instrument]_complete: Form completion status (0, 1, 2)
- redcap_url: REDCap instance URL
- project_url: Full project URL
"""
form_data = await request.form()
payload = dict(form_data)
logger.info(
"redcap_det_received",
project_id=payload.get("project_id"),
record=payload.get("record"),
instrument=payload.get("instrument"),
event=payload.get("redcap_event_name")
)
# Validate required fields
required = ["project_id", "record", "instrument"]
missing = [f for f in required if not payload.get(f)]
if missing:
logger.warning("redcap_det_missing_fields", missing=missing)
raise HTTPException(status_code=400, detail=f"Missing fields: {missing}")
# Queue background processing
background_tasks.add_task(
process_det_webhook,
db=db,
payload=payload
)
# REDCap expects immediate 200 response
return {"status": "accepted", "record": payload.get("record")}
async def process_det_webhook(db: AsyncSession, payload: dict):
"""Background task to process DET webhook."""
sync_service = REDCapSyncService(db)
try:
await sync_service.handle_det_trigger(
project_id=payload["project_id"],
record_id=payload["record"],
instrument=payload["instrument"],
event_name=payload.get("redcap_event_name"),
dag=payload.get("redcap_data_access_group")
)
except Exception as e:
logger.exception(
"redcap_det_processing_failed",
error=str(e),
payload=payload
)
1.2 Register Router¶
File: server/app/main.py (add import and include)
1.3 Webhook Security¶
Add HMAC signature verification for webhook authenticity.
File: server/app/config.py (add setting)
class Settings(BaseSettings):
# ... existing settings ...
# REDCap Webhook Security
REDCAP_WEBHOOK_SECRET: Optional[str] = None # For HMAC validation
REDCAP_WEBHOOK_IP_WHITELIST: Optional[str] = None # Comma-separated IPs
Phase 2: REDCap Sync Service¶
2.1 Service Implementation¶
File: server/app/services/redcap_sync.py
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.db.models import Study, Participant, ScheduledVisit, VisitWindow
from app.services.redcap import REDCapService
from app.services.scheduler import SchedulerService
from datetime import date, datetime
from typing import Optional
import structlog
logger = structlog.get_logger()
class REDCapSyncService:
"""
Service for synchronizing participants and data between REDCap and Metricis.
Handles:
- DET webhook processing
- Participant creation/updates from REDCap records
- Schedule creation based on enrollment date
- Pushing assessment data back to REDCap
"""
def __init__(self, db: AsyncSession):
self.db = db
async def handle_det_trigger(
self,
project_id: str,
record_id: str,
instrument: str,
event_name: Optional[str] = None,
dag: Optional[str] = None
):
"""
Process a Data Entry Trigger from REDCap.
Actions:
1. Find matching Metricis study by REDCap project ID
2. If enrollment form: sync participant and create schedule
3. If other form: update participant data if relevant
"""
# Find study by REDCap project ID
study = await self._find_study_by_redcap_project(project_id)
if not study:
logger.warning(
"redcap_det_unknown_project",
project_id=project_id
)
return
# Get study's REDCap configuration
redcap_config = await self._get_redcap_config(study.id)
if not redcap_config:
logger.warning(
"redcap_det_no_config",
study_id=study.id
)
return
# Check if this is the enrollment instrument
if instrument == redcap_config.get("enrollment_instrument"):
await self._sync_participant_from_redcap(
study=study,
record_id=record_id,
event_name=event_name,
dag=dag
)
else:
logger.debug(
"redcap_det_non_enrollment_form",
instrument=instrument,
record=record_id
)
async def _find_study_by_redcap_project(self, project_id: str) -> Optional[Study]:
"""Find Metricis study linked to REDCap project ID."""
result = await self.db.execute(
select(Study).where(
Study.redcap_enabled == True,
Study.metadata.contains({"redcap_project_id": project_id})
)
)
return result.scalar_one_or_none()
async def _get_redcap_config(self, study_id: int) -> Optional[dict]:
"""Get REDCap configuration for study."""
# This would query REDCapConfig model or study metadata
# Returns dict with: api_url, api_token, enrollment_instrument,
# enrollment_event, field_mapping, etc.
pass # Implementation depends on how config is stored
async def _sync_participant_from_redcap(
self,
study: Study,
record_id: str,
event_name: Optional[str],
dag: Optional[str]
):
"""
Sync a single participant from REDCap to Metricis.
1. Fetch full record from REDCap API
2. Extract participant fields per field mapping
3. Create or update Metricis participant
4. Create visit schedule if new participant
"""
redcap_config = await self._get_redcap_config(study.id)
redcap_service = REDCapService(
api_url=redcap_config["api_url"],
api_token=redcap_config["api_token"]
)
# Fetch record from REDCap
record_data = await redcap_service.export_records(
records=[record_id],
events=[event_name] if event_name else None
)
if not record_data:
logger.warning(
"redcap_sync_record_not_found",
record_id=record_id
)
return
record = record_data[0]
# Map REDCap fields to Metricis fields
field_mapping = redcap_config.get("field_mapping", {})
participant_data = self._map_redcap_to_participant(record, field_mapping)
# Check if participant already exists
existing = await self.db.execute(
select(Participant).where(
Participant.study_id == study.id,
Participant.external_id == record_id
)
)
participant = existing.scalar_one_or_none()
if participant:
# Update existing participant
for key, value in participant_data.items():
if value is not None:
setattr(participant, key, value)
logger.info(
"redcap_sync_participant_updated",
participant_id=participant.id,
record_id=record_id
)
else:
# Create new participant
participant = Participant(
study_id=study.id,
external_id=record_id,
status="enrolled",
**participant_data
)
self.db.add(participant)
await self.db.flush()
logger.info(
"redcap_sync_participant_created",
participant_id=participant.id,
record_id=record_id
)
# Create visit schedule for new participant
scheduler = SchedulerService(self.db)
await scheduler.schedule_visits_for_participant(participant.id)
await self.db.commit()
def _map_redcap_to_participant(
self,
record: dict,
field_mapping: dict
) -> dict:
"""
Map REDCap record fields to Metricis participant fields.
field_mapping example:
{
"email": "participant_email",
"phone": "mobile_phone",
"first_name": "first_name",
"last_name": "last_name",
"enrollment_date": "consent_date",
"date_of_birth": "dob"
}
"""
participant_data = {}
for metricis_field, redcap_field in field_mapping.items():
value = record.get(redcap_field)
if value:
# Handle date fields
if metricis_field in ("enrollment_date", "date_of_birth"):
try:
value = datetime.strptime(value, "%Y-%m-%d").date()
except ValueError:
value = None
participant_data[metricis_field] = value
return participant_data
async def push_assessment_to_redcap(
self,
session_id: int,
participant_id: int
):
"""
Push completed assessment data to REDCap.
1. Get session data and participant
2. Get REDCap configuration and field mapping
3. Transform Metricis data to REDCap format
4. Import to REDCap via API
5. Mark form as complete
"""
# Implementation builds on existing redcap.py import_cognitive_data()
pass
Phase 3: Database Model Changes¶
3.1 REDCap Configuration Model¶
Add or update model to store REDCap configuration per study.
File: server/app/db/models.py (additions)
class REDCapConfig(Base):
"""REDCap configuration for a study."""
__tablename__ = "redcap_configs"
id = Column(Integer, primary_key=True)
study_id = Column(Integer, ForeignKey("studies.id"), unique=True, nullable=False)
# Connection
api_url = Column(String, nullable=False)
api_token_encrypted = Column(String, nullable=False) # Encrypted storage
project_id = Column(String, nullable=True) # REDCap project ID (from API)
# Enrollment configuration
enrollment_instrument = Column(String, nullable=True) # Form that triggers sync
enrollment_event = Column(String, nullable=True) # Event for enrollment form
# Field mapping (Metricis field -> REDCap field)
participant_field_mapping = Column(JSON, default=dict)
# Event-Battery mapping (REDCap event -> Metricis battery)
event_battery_mapping = Column(JSON, default=dict)
# Example: {"baseline_arm_1": {"battery_id": 5, "target_day": 0}, ...}
# Data push configuration
data_field_mapping = Column(JSON, default=dict) # Assessment data -> REDCap fields
# Webhook
webhook_enabled = Column(Boolean, default=False)
webhook_secret = Column(String, nullable=True)
last_webhook_received = Column(DateTime, nullable=True)
# Sync status
last_sync_at = Column(DateTime, nullable=True)
sync_status = Column(String, default="pending") # pending, active, error
sync_error_message = Column(String, nullable=True)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
study = relationship("Study", back_populates="redcap_config")
3.2 Migration¶
File: server/alembic/versions/XXX_add_redcap_config_table.py
def upgrade():
op.create_table(
'redcap_configs',
sa.Column('id', sa.Integer(), primary_key=True),
sa.Column('study_id', sa.Integer(), sa.ForeignKey('studies.id'), unique=True, nullable=False),
sa.Column('api_url', sa.String(), nullable=False),
sa.Column('api_token_encrypted', sa.String(), nullable=False),
sa.Column('project_id', sa.String(), nullable=True),
sa.Column('enrollment_instrument', sa.String(), nullable=True),
sa.Column('enrollment_event', sa.String(), nullable=True),
sa.Column('participant_field_mapping', sa.JSON(), default=dict),
sa.Column('event_battery_mapping', sa.JSON(), default=dict),
sa.Column('data_field_mapping', sa.JSON(), default=dict),
sa.Column('webhook_enabled', sa.Boolean(), default=False),
sa.Column('webhook_secret', sa.String(), nullable=True),
sa.Column('last_webhook_received', sa.DateTime(), nullable=True),
sa.Column('last_sync_at', sa.DateTime(), nullable=True),
sa.Column('sync_status', sa.String(), default='pending'),
sa.Column('sync_error_message', sa.String(), nullable=True),
sa.Column('created_at', sa.DateTime(), default=datetime.utcnow),
sa.Column('updated_at', sa.DateTime(), default=datetime.utcnow),
)
def downgrade():
op.drop_table('redcap_configs')
Phase 4: Event-Battery Mapping¶
4.1 Configuration Schema¶
The event-battery mapping defines how REDCap events correspond to Metricis batteries and schedules.
{
"event_battery_mapping": {
"baseline_arm_1": {
"battery_id": 5,
"target_day": 0,
"window_before": 0,
"window_after": 7
},
"month3_arm_1": {
"battery_id": 5,
"target_day": 90,
"window_before": 7,
"window_after": 14
},
"month6_arm_1": {
"battery_id": 6,
"target_day": 180,
"window_before": 7,
"window_after": 14
},
"month12_arm_1": {
"battery_id": 6,
"target_day": 365,
"window_before": 14,
"window_after": 30
}
}
}
4.2 Schedule Generation from Mapping¶
Modify SchedulerService to use event-battery mapping when creating visits for REDCap-enabled studies.
File: server/app/services/scheduler.py (modification)
async def schedule_visits_for_participant(
self,
participant_id: int,
anchor_date: Optional[date] = None
) -> List[ScheduledVisit]:
"""
Create scheduled visits for a participant based on study configuration.
For REDCap-enabled studies: uses event_battery_mapping from REDCapConfig
For standard studies: uses VisitWindow definitions
"""
participant = await self._get_participant(participant_id)
study = await self._get_study(participant.study_id)
# Determine anchor date
if anchor_date is None:
anchor_date = participant.enrollment_date or date.today()
visits = []
if study.redcap_enabled:
# Use REDCap event-battery mapping
visits = await self._create_visits_from_redcap_mapping(
participant=participant,
anchor_date=anchor_date
)
else:
# Use standard visit windows
visits = await self._create_visits_from_windows(
participant=participant,
anchor_date=anchor_date
)
return visits
async def _create_visits_from_redcap_mapping(
self,
participant: Participant,
anchor_date: date
) -> List[ScheduledVisit]:
"""Create visits based on REDCap event-battery mapping."""
# Get REDCap config for study
result = await self.db.execute(
select(REDCapConfig).where(
REDCapConfig.study_id == participant.study_id
)
)
config = result.scalar_one_or_none()
if not config or not config.event_battery_mapping:
logger.warning(
"no_redcap_event_mapping",
study_id=participant.study_id
)
return []
visits = []
mapping = config.event_battery_mapping
for event_name, event_config in mapping.items():
target_date = anchor_date + timedelta(days=event_config["target_day"])
window_open = target_date - timedelta(days=event_config.get("window_before", 0))
window_close = target_date + timedelta(days=event_config.get("window_after", 7))
visit = ScheduledVisit(
participant_id=participant.id,
battery_id=event_config["battery_id"],
redcap_event_name=event_name,
target_date=target_date,
window_open=window_open,
window_close=window_close,
status="scheduled"
)
self.db.add(visit)
visits.append(visit)
await self.db.flush()
return visits
Phase 5: Data Push to REDCap¶
5.1 Assessment Data Mapping¶
Configure how Metricis assessment results map to REDCap form fields.
{
"data_field_mapping": {
"simple_rt": {
"redcap_instrument": "cognitive_assessments",
"fields": {
"simple_rt_mean": "srt_mean_rt",
"simple_rt_median": "srt_median_rt",
"simple_rt_std": "srt_sd_rt",
"simple_rt_accuracy": "srt_accuracy"
}
},
"digit_span": {
"redcap_instrument": "cognitive_assessments",
"fields": {
"digit_span_forward": "ds_forward_span",
"digit_span_backward": "ds_backward_span",
"digit_span_total": "ds_total_score"
}
},
"n_back": {
"redcap_instrument": "cognitive_assessments",
"fields": {
"n_back_accuracy": "nback_accuracy",
"n_back_dprime": "nback_dprime",
"n_back_mean_rt": "nback_mean_rt"
}
}
}
}
5.2 Push Implementation¶
File: server/app/services/redcap_sync.py (addition)
async def push_session_to_redcap(
self,
session_id: int
) -> dict:
"""
Push completed session data to REDCap.
Returns:
dict with status and any errors
"""
# Get session with participant and study
session = await self._get_session_with_relations(session_id)
if not session:
return {"status": "error", "message": "Session not found"}
participant = session.participant
study = participant.study
if not study.redcap_enabled:
return {"status": "skipped", "message": "Study not REDCap-enabled"}
# Get REDCap config
config = await self._get_redcap_config(study.id)
if not config:
return {"status": "error", "message": "No REDCap configuration"}
# Get the scheduled visit for this session
visit = await self._get_visit_for_session(session_id)
if not visit or not visit.redcap_event_name:
return {"status": "error", "message": "No REDCap event for visit"}
# Build REDCap import data
import_data = self._build_redcap_import(
session=session,
record_id=participant.external_id,
event_name=visit.redcap_event_name,
field_mapping=config.data_field_mapping
)
# Push to REDCap
redcap_service = REDCapService(
api_url=config.api_url,
api_token=self._decrypt_token(config.api_token_encrypted)
)
result = await redcap_service.import_records(
records=import_data,
overwrite_behavior="overwrite"
)
# Update sync status
visit.redcap_sync_status = "synced"
visit.redcap_synced_at = datetime.utcnow()
await self.db.commit()
logger.info(
"redcap_data_pushed",
session_id=session_id,
record_id=participant.external_id,
event=visit.redcap_event_name
)
return {"status": "success", "records_imported": result.get("count", 1)}
def _build_redcap_import(
self,
session: Session,
record_id: str,
event_name: str,
field_mapping: dict
) -> List[dict]:
"""
Transform Metricis session data to REDCap import format.
"""
record = {
"record_id": record_id,
"redcap_event_name": event_name
}
# Get task summaries from session
summaries = session.task_summaries or {}
for task_name, task_config in field_mapping.items():
if task_name not in summaries:
continue
task_data = summaries[task_name]
instrument = task_config.get("redcap_instrument")
fields = task_config.get("fields", {})
for metricis_field, redcap_field in fields.items():
value = task_data.get(metricis_field)
if value is not None:
# Format numeric values
if isinstance(value, float):
value = round(value, 2)
record[redcap_field] = value
# Mark instrument as complete
if instrument:
record[f"{instrument}_complete"] = "2" # 2 = Complete
return [record]
Phase 6: API Endpoints for Configuration¶
6.1 REDCap Config Endpoints¶
File: server/app/routers/redcap.py (additions)
@router.get("/config/{study_id}/event-mapping")
async def get_event_battery_mapping(
study_id: int,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Get event-to-battery mapping for a REDCap-enabled study."""
config = await get_redcap_config(db, study_id)
if not config:
raise HTTPException(404, "REDCap config not found")
return {
"study_id": study_id,
"event_battery_mapping": config.event_battery_mapping,
"enrollment_instrument": config.enrollment_instrument,
"enrollment_event": config.enrollment_event
}
@router.put("/config/{study_id}/event-mapping")
async def update_event_battery_mapping(
study_id: int,
mapping: EventBatteryMappingRequest,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Update event-to-battery mapping."""
config = await get_redcap_config(db, study_id)
if not config:
raise HTTPException(404, "REDCap config not found")
config.event_battery_mapping = mapping.event_battery_mapping
await db.commit()
return {"status": "updated"}
@router.post("/config/{study_id}/test-connection")
async def test_redcap_connection(
study_id: int,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Test REDCap API connection and fetch project info."""
config = await get_redcap_config(db, study_id)
if not config:
raise HTTPException(404, "REDCap config not found")
try:
redcap_service = REDCapService(
api_url=config.api_url,
api_token=decrypt_token(config.api_token_encrypted)
)
project_info = await redcap_service.export_project_info()
# Update stored project ID
config.project_id = str(project_info.get("project_id"))
config.sync_status = "active"
config.sync_error_message = None
await db.commit()
return {
"status": "connected",
"project_title": project_info.get("project_title"),
"project_id": project_info.get("project_id"),
"is_longitudinal": project_info.get("is_longitudinal"),
"record_count": project_info.get("record_count")
}
except Exception as e:
config.sync_status = "error"
config.sync_error_message = str(e)
await db.commit()
raise HTTPException(400, f"Connection failed: {str(e)}")
@router.get("/config/{study_id}/instruments")
async def get_redcap_instruments(
study_id: int,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Fetch available instruments from REDCap project."""
config = await get_redcap_config(db, study_id)
if not config:
raise HTTPException(404, "REDCap config not found")
redcap_service = REDCapService(
api_url=config.api_url,
api_token=decrypt_token(config.api_token_encrypted)
)
instruments = await redcap_service.export_instruments()
return {"instruments": instruments}
@router.get("/config/{study_id}/events")
async def get_redcap_events(
study_id: int,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""Fetch available events from REDCap project (longitudinal only)."""
config = await get_redcap_config(db, study_id)
if not config:
raise HTTPException(404, "REDCap config not found")
redcap_service = REDCapService(
api_url=config.api_url,
api_token=decrypt_token(config.api_token_encrypted)
)
events = await redcap_service.export_events()
return {"events": events}
6.2 Pydantic Schemas¶
File: server/app/schemas/redcap.py (additions)
from pydantic import BaseModel
from typing import Dict, Optional
class EventBatteryConfig(BaseModel):
battery_id: int
target_day: int
window_before: int = 0
window_after: int = 7
class EventBatteryMappingRequest(BaseModel):
event_battery_mapping: Dict[str, EventBatteryConfig]
class REDCapConfigResponse(BaseModel):
id: int
study_id: int
api_url: str
project_id: Optional[str]
enrollment_instrument: Optional[str]
enrollment_event: Optional[str]
participant_field_mapping: dict
event_battery_mapping: dict
webhook_enabled: bool
sync_status: str
class Config:
from_attributes = True
Phase 7: Portal UI Components¶
7.1 REDCap Event Mapping Configuration¶
Add UI for configuring event-battery mapping in the portal.
File: portal/src/pages/admin/REDCapConfig.tsx (additions)
// Add section for event-battery mapping
interface EventBatteryMapping {
[eventName: string]: {
battery_id: number;
target_day: number;
window_before: number;
window_after: number;
};
}
function EventBatteryMappingSection({ studyId }: { studyId: number }) {
const [mapping, setMapping] = useState<EventBatteryMapping>({});
const [events, setEvents] = useState<string[]>([]);
const [batteries, setBatteries] = useState<Battery[]>([]);
// Fetch REDCap events and available batteries
useEffect(() => {
fetchRedcapEvents(studyId).then(setEvents);
fetchBatteries(studyId).then(setBatteries);
fetchEventMapping(studyId).then(setMapping);
}, [studyId]);
const handleMappingChange = (eventName: string, field: string, value: any) => {
setMapping(prev => ({
...prev,
[eventName]: {
...prev[eventName],
[field]: value
}
}));
};
return (
<section className="config-section">
<h3>Event-Battery Mapping</h3>
<p className="help-text">
Map REDCap events to Metricis batteries. Each event will trigger
the specified battery at the target day after enrollment.
</p>
<table className="mapping-table">
<thead>
<tr>
<th>REDCap Event</th>
<th>Battery</th>
<th>Target Day</th>
<th>Window (days before/after)</th>
</tr>
</thead>
<tbody>
{events.map(event => (
<tr key={event}>
<td>{event}</td>
<td>
<select
value={mapping[event]?.battery_id || ''}
onChange={e => handleMappingChange(event, 'battery_id', parseInt(e.target.value))}
>
<option value="">-- Select Battery --</option>
{batteries.map(b => (
<option key={b.id} value={b.id}>{b.name}</option>
))}
</select>
</td>
<td>
<input
type="number"
value={mapping[event]?.target_day || 0}
onChange={e => handleMappingChange(event, 'target_day', parseInt(e.target.value))}
/>
</td>
<td>
<input
type="number"
value={mapping[event]?.window_before || 0}
onChange={e => handleMappingChange(event, 'window_before', parseInt(e.target.value))}
style={{ width: '60px' }}
/> /
<input
type="number"
value={mapping[event]?.window_after || 7}
onChange={e => handleMappingChange(event, 'window_after', parseInt(e.target.value))}
style={{ width: '60px' }}
/>
</td>
</tr>
))}
</tbody>
</table>
<button onClick={() => saveEventMapping(studyId, mapping)}>
Save Mapping
</button>
</section>
);
}
Phase 8: Implementation Sequence¶
Recommended Order¶
- Phase 3: Database Models (1-2 days)
- Create REDCapConfig model and migration
-
Update Study relationship
-
Phase 1: Webhook Endpoint (1-2 days)
- Create webhook router
- Add security validation
-
Basic logging
-
Phase 2: Sync Service (2-3 days)
- Participant sync from REDCap
- Field mapping logic
-
Error handling
-
Phase 4: Event-Battery Mapping (1-2 days)
- Modify scheduler service
-
Create visits from mapping
-
Phase 5: Data Push (2-3 days)
- Assessment data transformation
- REDCap import with form completion
-
Sync status tracking
-
Phase 6: API Endpoints (1-2 days)
- Configuration endpoints
- Test connection endpoint
-
Fetch instruments/events
-
Phase 7: Portal UI (2-3 days)
- Event mapping configuration
- Field mapping interface
- Webhook status display
Total Estimated Effort: 10-17 days¶
Testing Strategy¶
Unit Tests¶
# test_redcap_sync.py
async def test_det_webhook_creates_participant():
"""DET webhook should create new participant."""
pass
async def test_det_webhook_updates_existing_participant():
"""DET webhook should update existing participant."""
pass
async def test_schedule_created_from_event_mapping():
"""Visits should be created per event-battery mapping."""
pass
async def test_data_push_transforms_correctly():
"""Assessment data should transform to REDCap format."""
pass
Integration Tests¶
- Mock REDCap API responses
- Test full webhook → participant → schedule flow
- Test data push with form completion marking
Manual Testing¶
- Configure test REDCap project with DET pointing to local Metricis
- Create record in REDCap → verify participant appears in Metricis
- Complete assessment → verify data appears in REDCap
Security Considerations¶
- Token Storage: API tokens must be encrypted at rest
- Webhook Authentication: Implement HMAC signature verification
- IP Whitelisting: Optionally restrict webhook source IPs
- Rate Limiting: Apply rate limits to webhook endpoint
- Audit Logging: Log all sync operations for compliance
Rollback Plan¶
If issues arise:
- Disable webhook in REDCap project settings
- Disable
webhook_enabledflag in Metricis config - Participants can be manually imported via CSV as fallback
- Data can be manually exported and imported to REDCap
Open Questions¶
- Token Encryption: Use Fernet or AWS KMS for token encryption?
- Retry Logic: How many times to retry failed syncs?
- Conflict Resolution: If participant exists in both systems with different data, which wins?
- Multi-arm Studies: How to handle participants moving between arms?
References¶
- REDCap API Documentation
- Data Entry Triggers Documentation
- PyCap Library
- Existing Metricis REDCap integration:
server/app/services/redcap.py