apunkt/SurfSense
1
0
Fork 0
mirror of https://github.com/MODSetter/SurfSense.git synced 2026-07-22 23:31:12 +02:00
Code Issues Projects Releases Packages Wiki Activity

Merge remote changes and fix date range filtering in connectors

Browse source 
- Fixed Jira connector to use constructed JQL query for date filtering
- Fixed ClickUp connector to include date range parameters in API request
- Resolved merge conflicts with remote branch improvements
- Enhanced ClickUp date handling with complete day coverage (00:00:00 to 23:59:59)
This commit is contained in:
Aditya Vaish 2025-10-16 10:12:56 +05:30
commit b07ce72f05
18 changed files with 19692 additions and 11296 deletions
Download patch file Download diff file Expand all files Collapse all files

3
.idea/.gitignore generated vendored Normal file
View file

@ -0,0 +1,3 @@
# Default ignored files
/shelf/
/workspace.xml

8
.idea/SurfSense.iml generated Normal file
View file

@ -0,0 +1,8 @@
<?xml version="1.0" encoding="UTF-8"?>
<module type="PYTHON_MODULE" version="4">
<component name="NewModuleRootManager">
<content url="file://$MODULE_DIR$" />
<orderEntry type="inheritedJdk" />
<orderEntry type="sourceFolder" forTests="false" />
</component>
</module>

6
.idea/inspectionProfiles/profiles_settings.xml generated Normal file
View file

@ -0,0 +1,6 @@
<component name="InspectionProjectProfileManager">
<settings>
<option name="USE_PROJECT_PROFILE" value="false" />
<version value="1.0" />
</settings>
</component>

8
.idea/modules.xml generated Normal file
View file

@ -0,0 +1,8 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="ProjectModuleManager">
<modules>
<module fileurl="file://$PROJECT_DIR$/.idea/SurfSense.iml" filepath="$PROJECT_DIR$/.idea/SurfSense.iml" />
</modules>
</component>
</project>

6
.idea/vcs.xml generated Normal file
View file

@ -0,0 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="VcsDirectoryMappings">
<mapping directory="" vcs="Git" />
</component>
</project>

100
surfsense_backend/alembic/versions/23_add_connector_schedules_table.py Normal file
View file

@ -0,0 +1,100 @@
"""Add connector schedules table
Revision ID: 23
Revises: 22
"""
from collections.abc import Sequence
from sqlalchemy import inspect
from alembic import op
# revision identifiers, used by Alembic.
revision: str = "23"
down_revision: str | None = "22"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
"""Upgrade schema - add ScheduleType enum and connector_schedules table."""
# Create ScheduleType enum if it doesn't exist
op.execute(
"""
DO $$
BEGIN
IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'scheduletype') THEN
CREATE TYPE scheduletype AS ENUM ('HOURLY', 'DAILY', 'WEEKLY', 'CUSTOM');
END IF;
END$$;
"""
)
# Create connector_schedules table if it doesn't exist
op.execute(
"""
CREATE TABLE IF NOT EXISTS connector_schedules (
id SERIAL PRIMARY KEY,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
connector_id INTEGER NOT NULL REFERENCES search_source_connectors(id) ON DELETE CASCADE,
search_space_id INTEGER NOT NULL REFERENCES searchspaces(id) ON DELETE CASCADE,
schedule_type scheduletype NOT NULL,
cron_expression VARCHAR(100),
daily_time TIME,
weekly_day SMALLINT,
weekly_time TIME,
hourly_minute SMALLINT,
is_active BOOLEAN NOT NULL DEFAULT TRUE,
last_run_at TIMESTAMPTZ,
next_run_at TIMESTAMPTZ,
CONSTRAINT uq_connector_search_space UNIQUE (connector_id, search_space_id)
);
"""
)
# Get existing indexes
conn = op.get_bind()
inspector = inspect(conn)
existing_indexes = [
idx["name"] for idx in inspector.get_indexes("connector_schedules")
]
# Create indexes only if they don't already exist
if "ix_connector_schedules_id" not in existing_indexes:
op.create_index("ix_connector_schedules_id", "connector_schedules", ["id"])
if "ix_connector_schedules_created_at" not in existing_indexes:
op.create_index(
"ix_connector_schedules_created_at", "connector_schedules", ["created_at"]
)
if "ix_connector_schedules_connector_id" not in existing_indexes:
op.create_index(
"ix_connector_schedules_connector_id", "connector_schedules", ["connector_id"]
)
if "ix_connector_schedules_is_active" not in existing_indexes:
op.create_index(
"ix_connector_schedules_is_active", "connector_schedules", ["is_active"]
)
if "ix_connector_schedules_next_run_at" not in existing_indexes:
op.create_index(
"ix_connector_schedules_next_run_at", "connector_schedules", ["next_run_at"]
)
def downgrade() -> None:
"""Downgrade schema - remove connector_schedules table and enum."""
# Drop indexes
op.drop_index("ix_connector_schedules_next_run_at", table_name="connector_schedules")
op.drop_index("ix_connector_schedules_is_active", table_name="connector_schedules")
op.drop_index("ix_connector_schedules_connector_id", table_name="connector_schedules")
op.drop_index("ix_connector_schedules_created_at", table_name="connector_schedules")
op.drop_index("ix_connector_schedules_id", table_name="connector_schedules")
# Drop table
op.drop_table("connector_schedules")
# Drop enum
op.execute("DROP TYPE IF EXISTS scheduletype")

38
surfsense_backend/app/app.py
View file

@ -1,4 +1,6 @@
from contextlib import asynccontextmanager
import asyncio
import logging
from contextlib import asynccontextmanager, suppress
from fastapi import Depends, FastAPI
from fastapi.middleware.cors import CORSMiddleware
@ -7,15 +9,45 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.config import config
from app.db import User, create_db_and_tables, get_async_session
from app.routes import router as crud_router
from app.routes.connector_schedules_routes import router as connector_schedules_router
from app.routes.scheduler_routes import router as scheduler_router
from app.schemas import UserCreate, UserRead, UserUpdate
from app.services.connector_scheduler_service import start_scheduler, stop_scheduler
from app.users import SECRET, auth_backend, current_active_user, fastapi_users
logger = logging.getLogger(__name__)
@asynccontextmanager
async def lifespan(app: FastAPI):
# Not needed if you setup a migration system like Alembic
"""Application lifespan manager with scheduler integration."""
# Startup
logger.info("Starting SurfSense application...")
# Create database tables
await create_db_and_tables()
logger.info("Database tables created/verified")
# Start the connector scheduler service
scheduler_task = asyncio.create_task(start_scheduler())
logger.info("Connector scheduler service started")
yield
# Shutdown
logger.info("Shutting down SurfSense application...")
# Stop the scheduler service
await stop_scheduler()
logger.info("Connector scheduler service stopped")
# Cancel the scheduler task
if not scheduler_task.done():
scheduler_task.cancel()
with suppress(asyncio.CancelledError):
await scheduler_task
logger.info("Application shutdown complete")
app = FastAPI(lifespan=lifespan)
@ -65,6 +97,8 @@ if config.AUTH_TYPE == "GOOGLE":
)
app.include_router(crud_router, prefix="/api/v1", tags=["crud"])
app.include_router(connector_schedules_router, prefix="/api/v1", tags=["connector-schedules"])
app.include_router(scheduler_router, prefix="/api/v1", tags=["scheduler"])
@app.get("/verify-token")

15
surfsense_backend/app/connectors/clickup_connector.py
View file

@ -169,9 +169,16 @@ class ClickUpConnector:
Tuple containing (tasks list, error message or None)
"""
try:
# Convert dates to Unix timestamps (milliseconds)
start_timestamp = int(datetime.strptime(start_date, "%Y-%m-%d").timestamp() * 1000)
end_timestamp = int(datetime.strptime(end_date, "%Y-%m-%d").timestamp() * 1000)
# Convert date strings to Unix timestamps (milliseconds)
start_datetime = datetime.strptime(start_date, "%Y-%m-%d")
end_datetime = datetime.strptime(end_date, "%Y-%m-%d")
# Set time to start and end of day for complete coverage
start_datetime = start_datetime.replace(hour=0, minute=0, second=0, microsecond=0)
end_datetime = end_datetime.replace(hour=23, minute=59, second=59, microsecond=999999)
start_timestamp = int(start_datetime.timestamp() * 1000)
end_timestamp = int(end_datetime.timestamp() * 1000)
params = {
"page": 0,
@ -179,7 +186,7 @@ class ClickUpConnector:
"reverse": "true",
"subtasks": "true",
"include_closed": str(include_closed).lower(),
# Date filtering parameters
# Date filtering - filter by both created and updated dates
"date_created_gt": start_timestamp,
"date_created_lt": end_timestamp,
"date_updated_gt": start_timestamp,

2
surfsense_backend/app/connectors/jira_connector.py
View file

@ -229,7 +229,7 @@ class JiraConnector:
_jql = f"{date_filter} ORDER BY created DESC"
if project_key:
_jql = (
f'project = "{project_key}" AND {date_filter} ORDER BY created DESC'
f'project = "{project_key}" AND ({date_filter}) ORDER BY created DESC'
)
# Define fields to retrieve

65
surfsense_backend/app/db.py
View file

@ -16,6 +16,7 @@ from sqlalchemy import (
Integer,
String,
Text,
Time,
UniqueConstraint,
text,
)
@ -129,6 +130,13 @@ class LogStatus(str, Enum):
FAILED = "FAILED"
class ScheduleType(str, Enum):
HOURLY = "HOURLY"
DAILY = "DAILY"
WEEKLY = "WEEKLY"
CUSTOM = "CUSTOM"
class Base(DeclarativeBase):
pass
@ -246,21 +254,10 @@ class SearchSpace(BaseModel, TimestampMixin):
order_by="Log.id",
cascade="all, delete-orphan",
)
search_source_connectors = relationship(
"SearchSourceConnector",
back_populates="search_space",
order_by="SearchSourceConnector.id",
cascade="all, delete-orphan",
)
llm_configs = relationship(
"LLMConfig",
back_populates="search_space",
order_by="LLMConfig.id",
cascade="all, delete-orphan",
)
user_preferences = relationship(
"UserSearchSpacePreference",
connector_schedules = relationship(
"ConnectorSchedule",
back_populates="search_space",
order_by="ConnectorSchedule.id",
cascade="all, delete-orphan",
)
@ -292,6 +289,46 @@ class SearchSourceConnector(BaseModel, TimestampMixin):
user_id = Column(
UUID(as_uuid=True), ForeignKey("user.id", ondelete="CASCADE"), nullable=False
)
user = relationship("User", back_populates="search_source_connectors")
schedules = relationship(
"ConnectorSchedule",
back_populates="connector",
cascade="all, delete-orphan",
)
class ConnectorSchedule(BaseModel, TimestampMixin):
__tablename__ = "connector_schedules"
__table_args__ = (
UniqueConstraint(
"connector_id", "search_space_id", name="uq_connector_search_space"
),
)
connector_id = Column(
Integer,
ForeignKey("search_source_connectors.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
search_space_id = Column(
Integer,
ForeignKey("searchspaces.id", ondelete="CASCADE"),
nullable=False,
)
schedule_type = Column(SQLAlchemyEnum(ScheduleType), nullable=False)
cron_expression = Column(String(100), nullable=True)
# Optional schedule config fields (persist user selections)
daily_time = Column(Time(timezone=False), nullable=True)
weekly_day = Column(Integer, nullable=True) # 0=Mon .. 6=Sun
weekly_time = Column(Time(timezone=False), nullable=True)
hourly_minute = Column(Integer, nullable=True) # 0..59
is_active = Column(Boolean, nullable=False, default=True, index=True)
last_run_at = Column(TIMESTAMP(timezone=True), nullable=True)
next_run_at = Column(TIMESTAMP(timezone=True), nullable=True, index=True)
connector = relationship("SearchSourceConnector", back_populates="schedules")
search_space = relationship("SearchSpace", back_populates="connector_schedules")
class LLMConfig(BaseModel, TimestampMixin):

434
surfsense_backend/app/routes/connector_schedules_routes.py Normal file
View file

@ -0,0 +1,434 @@
"""
ConnectorSchedule routes for CRUD operations:
POST /connector-schedules/ - Create a new schedule
GET /connector-schedules/ - List all schedules for the current user
GET /connector-schedules/{schedule_id} - Get a specific schedule
PUT /connector-schedules/{schedule_id} - Update a specific schedule
DELETE /connector-schedules/{schedule_id} - Delete a specific schedule
PATCH /connector-schedules/{schedule_id}/toggle - Activate/deactivate a schedule
"""
import logging
from datetime import timezone
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from app.db import (
ConnectorSchedule,
ScheduleType,
SearchSourceConnector,
SearchSpace,
User,
get_async_session,
)
from app.schemas import (
ConnectorScheduleCreate,
ConnectorScheduleRead,
ConnectorScheduleUpdate,
)
from app.users import current_active_user
from app.utils.check_ownership import check_ownership
from app.utils.schedule_helpers import calculate_next_run
# Set up logging
logger = logging.getLogger(__name__)
router = APIRouter()
@router.post("/connector-schedules/", response_model=ConnectorScheduleRead)
async def create_connector_schedule(
schedule: ConnectorScheduleCreate,
session: AsyncSession = Depends(get_async_session),
user: User = Depends(current_active_user),
):
"""
Create a new connector schedule.
Each connector can have only one schedule per search space.
The schedule will automatically calculate the next run time based on the schedule type.
"""
try:
# Verify connector belongs to user
connector = await check_ownership(
session, SearchSourceConnector, schedule.connector_id, user
)
# Verify connector is indexable
if not connector.is_indexable:
raise HTTPException(
status_code=400,
detail=f"Connector {connector.name} is not indexable and cannot be scheduled",
)
# Verify search space belongs to user
await check_ownership(session, SearchSpace, schedule.search_space_id, user)
# Ensure the connector belongs to the same search space
if connector.search_space_id != schedule.search_space_id:
raise HTTPException(
status_code=400,
detail="Connector does not belong to the provided search space",
)
# Check if schedule already exists for this connector-space pair
result = await session.execute(
select(ConnectorSchedule).filter(
ConnectorSchedule.connector_id == schedule.connector_id,
ConnectorSchedule.search_space_id == schedule.search_space_id,
)
)
existing_schedule = result.scalars().first()
if existing_schedule:
raise HTTPException(
status_code=409,
detail=f"A schedule already exists for connector {schedule.connector_id} and search space {schedule.search_space_id}",
)
# Calculate next run time only if schedule is active
next_run_at = None
if schedule.is_active:
next_run_at = calculate_next_run(
schedule.schedule_type,
schedule.cron_expression,
schedule.daily_time,
schedule.weekly_day,
schedule.weekly_time,
schedule.hourly_minute
)
# Create schedule (only DB columns)
db_data = schedule.model_dump(
include={
"connector_id",
"search_space_id",
"schedule_type",
"cron_expression",
"daily_time",
"weekly_day",
"weekly_time",
"hourly_minute",
"is_active"
}
)
db_schedule = ConnectorSchedule(**db_data, next_run_at=next_run_at)
session.add(db_schedule)
await session.commit()
await session.refresh(db_schedule)
logger.info(
f"Created schedule {db_schedule.id} for connector {schedule.connector_id} (next run: {next_run_at})"
)
return db_schedule
except HTTPException:
await session.rollback()
raise
except IntegrityError as e:
await session.rollback()
raise HTTPException(
status_code=409,
detail=f"A schedule already exists for this connector and search space: {e!s}",
) from e
except Exception as e:
await session.rollback()
logger.error(f"Failed to create connector schedule: {e!s}")
raise HTTPException(
status_code=500, detail=f"Failed to create connector schedule: {e!s}"
) from e
@router.get("/connector-schedules/", response_model=list[ConnectorScheduleRead])
async def read_connector_schedules(
skip: int = 0,
limit: int = 100,
connector_id: int | None = None,
search_space_id: int | None = None,
is_active: bool | None = None,
session: AsyncSession = Depends(get_async_session),
user: User = Depends(current_active_user),
):
"""
List all connector schedules for the current user.
Optional filters:
- connector_id: Filter by specific connector
- search_space_id: Filter by specific search space
- is_active: Filter by active/inactive status
"""
try:
# Build query to get schedules for connectors owned by user
query = (
select(ConnectorSchedule)
.join(SearchSourceConnector)
.filter(SearchSourceConnector.user_id == user.id)
)
# Apply filters
if connector_id is not None:
query = query.filter(ConnectorSchedule.connector_id == connector_id)
if search_space_id is not None:
query = query.filter(ConnectorSchedule.search_space_id == search_space_id)
if is_active is not None:
query = query.filter(ConnectorSchedule.is_active == is_active)
result = await session.execute(query.offset(skip).limit(limit))
return result.scalars().all()
except Exception as e:
logger.error(f"Failed to fetch connector schedules: {e!s}")
raise HTTPException(
status_code=500, detail=f"Failed to fetch connector schedules: {e!s}"
) from e
@router.get("/connector-schedules/{schedule_id}", response_model=ConnectorScheduleRead)
async def read_connector_schedule(
schedule_id: int,
session: AsyncSession = Depends(get_async_session),
user: User = Depends(current_active_user),
):
"""Get a specific connector schedule by ID."""
try:
# Get schedule
result = await session.execute(
select(ConnectorSchedule).filter(ConnectorSchedule.id == schedule_id)
)
schedule = result.scalars().first()
if not schedule:
raise HTTPException(status_code=404, detail="Schedule not found")
# Verify schedule's connector belongs to user
await check_ownership(session, SearchSourceConnector, schedule.connector_id, user)
return schedule
except HTTPException:
raise
except Exception as e:
logger.error(f"Failed to fetch connector schedule: {e!s}")
raise HTTPException(
status_code=500, detail=f"Failed to fetch connector schedule: {e!s}"
) from e
@router.put("/connector-schedules/{schedule_id}", response_model=ConnectorScheduleRead)
async def update_connector_schedule(
schedule_id: int,
schedule_update: ConnectorScheduleUpdate,
session: AsyncSession = Depends(get_async_session),
user: User = Depends(current_active_user),
):
"""
Update a connector schedule.
Can update schedule_type, cron_expression, and is_active.
If schedule_type changes, next_run_at is recalculated automatically.
"""
try:
# Get the existing schedule
result = await session.execute(
select(ConnectorSchedule).filter(ConnectorSchedule.id == schedule_id)
)
schedule = result.scalars().first()
if not schedule:
raise HTTPException(status_code=404, detail="Schedule not found")
# Verify schedule's connector belongs to user
await check_ownership(session, SearchSourceConnector, schedule.connector_id, user)
# Update fields that were provided
update_data = schedule_update.model_dump(exclude_unset=True)
# Effective types/values after this update (for validation and next-run calc)
effective_type = update_data.get("schedule_type", schedule.schedule_type)
effective_cron = update_data.get("cron_expression", schedule.cron_expression)
effective_daily = update_data.get("daily_time", schedule.daily_time)
effective_weekly_day = update_data.get("weekly_day", schedule.weekly_day)
effective_weekly_time = update_data.get("weekly_time", schedule.weekly_time)
effective_hourly_minute = update_data.get("hourly_minute", schedule.hourly_minute)
will_be_active = update_data.get("is_active", schedule.is_active)
# Disallow cron for non-CUSTOM schedules
if "cron_expression" in update_data and effective_type != ScheduleType.CUSTOM:
raise HTTPException(
status_code=400,
detail="cron_expression is only valid when schedule_type is CUSTOM",
)
# Check if any time-related fields changed (requiring next_run recalc)
time_fields_changed = any(
field in update_data
for field in ["daily_time", "weekly_day", "weekly_time", "hourly_minute"]
)
# If schedule_type is being updated or time fields changed, recalculate next_run_at
if "schedule_type" in update_data or time_fields_changed:
if will_be_active:
update_data["next_run_at"] = calculate_next_run(
effective_type,
effective_cron,
effective_daily,
effective_weekly_day,
effective_weekly_time,
effective_hourly_minute,
)
# Ensure timezone-aware UTC
if update_data["next_run_at"].tzinfo is None:
update_data["next_run_at"] = update_data["next_run_at"].replace(tzinfo=timezone.utc)
else:
update_data["next_run_at"] = None
elif "cron_expression" in update_data and schedule.schedule_type == ScheduleType.CUSTOM:
# If only cron_expression is updated for CUSTOM schedule, recalculate next_run_at
if will_be_active:
update_data["next_run_at"] = calculate_next_run(
schedule.schedule_type,
update_data["cron_expression"],
schedule.daily_time,
schedule.weekly_day,
schedule.weekly_time,
schedule.hourly_minute,
)
# Ensure timezone-aware UTC
if update_data["next_run_at"].tzinfo is None:
update_data["next_run_at"] = update_data["next_run_at"].replace(tzinfo=timezone.utc)
else:
update_data["next_run_at"] = None
# Handle activation toggles
if "is_active" in update_data:
if update_data["is_active"] is False:
update_data["next_run_at"] = None
elif update_data["is_active"] is True and "next_run_at" not in update_data:
# Schedule is being activated and next_run_at wasn't already calculated above
next_run = calculate_next_run(
effective_type,
effective_cron,
effective_daily,
effective_weekly_day,
effective_weekly_time,
effective_hourly_minute,
)
if next_run.tzinfo is None:
next_run = next_run.replace(tzinfo=timezone.utc)
update_data["next_run_at"] = next_run
# Apply updates
for field, value in update_data.items():
setattr(schedule, field, value)
await session.commit()
await session.refresh(schedule)
logger.info(f"Updated schedule {schedule_id} with fields: {list(update_data.keys())}")
return schedule
except HTTPException:
await session.rollback()
raise
except Exception as e:
await session.rollback()
logger.error(f"Failed to update connector schedule: {e!s}")
raise HTTPException(
status_code=500, detail=f"Failed to update connector schedule: {e!s}"
) from e
@router.delete("/connector-schedules/{schedule_id}")
async def delete_connector_schedule(
schedule_id: int,
session: AsyncSession = Depends(get_async_session),
user: User = Depends(current_active_user),
):
"""Delete a connector schedule."""
try:
# Get the existing schedule
result = await session.execute(
select(ConnectorSchedule).filter(ConnectorSchedule.id == schedule_id)
)
schedule = result.scalars().first()
if not schedule:
raise HTTPException(status_code=404, detail="Schedule not found")
# Verify schedule's connector belongs to user
await check_ownership(session, SearchSourceConnector, schedule.connector_id, user)
# Delete the schedule
await session.delete(schedule)
await session.commit()
logger.info(f"Deleted schedule {schedule_id}")
return {"message": "Schedule deleted successfully"}
except HTTPException:
await session.rollback()
raise
except Exception as e:
await session.rollback()
logger.error(f"Failed to delete connector schedule: {e!s}")
raise HTTPException(
status_code=500, detail=f"Failed to delete connector schedule: {e!s}"
) from e
@router.patch("/connector-schedules/{schedule_id}/toggle", response_model=ConnectorScheduleRead)
async def toggle_connector_schedule(
schedule_id: int,
session: AsyncSession = Depends(get_async_session),
user: User = Depends(current_active_user),
):
"""Toggle the active status of a connector schedule."""
try:
# Get the existing schedule
result = await session.execute(
select(ConnectorSchedule).filter(ConnectorSchedule.id == schedule_id)
)
schedule = result.scalars().first()
if not schedule:
raise HTTPException(status_code=404, detail="Schedule not found")
# Verify schedule's connector belongs to user
await check_ownership(session, SearchSourceConnector, schedule.connector_id, user)
# Toggle the active status
schedule.is_active = not schedule.is_active
# Update next_run_at based on new active status
if schedule.is_active:
# Schedule is being activated, calculate next_run_at if it's None
if schedule.next_run_at is None:
schedule.next_run_at = calculate_next_run(
schedule.schedule_type,
schedule.cron_expression,
schedule.daily_time,
schedule.weekly_day,
schedule.weekly_time,
schedule.hourly_minute
)
else:
# Schedule is being deactivated, set next_run_at to None
schedule.next_run_at = None
await session.commit()
await session.refresh(schedule)
logger.info(f"Toggled schedule {schedule_id} to {'active' if schedule.is_active else 'inactive'}")
return schedule
except HTTPException:
await session.rollback()
raise
except Exception as e:
await session.rollback()
logger.error(f"Failed to toggle connector schedule: {e!s}")
raise HTTPException(
status_code=500, detail=f"Failed to toggle connector schedule: {e!s}"
) from e

204
surfsense_backend/app/routes/scheduler_routes.py Normal file
View file

@ -0,0 +1,204 @@
"""
Scheduler management routes for monitoring and controlling the connector scheduler service.
"""
import logging
from typing import Dict, Any
from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks
from sqlalchemy import select, Integer
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from app.db import get_async_session, Log
from app.schemas import ConnectorScheduleRead
from app.services.connector_scheduler_service import get_scheduler
from app.users import User, current_active_user
logger = logging.getLogger(__name__)
router = APIRouter()
@router.get("/scheduler/status", response_model=Dict[str, Any])
async def get_scheduler_status(
user: User = Depends(current_active_user),
):
"""
Get the current status of the connector scheduler service.
Returns information about:
- Whether the scheduler is running
- Number of active jobs
- Configuration details
- Active job details
"""
try:
scheduler = await get_scheduler()
status = await scheduler.get_scheduler_status()
return status
except Exception as e:
logger.error(f"Error getting scheduler status: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to get scheduler status: {e}"
) from e
@router.post("/scheduler/schedules/{schedule_id}/force-execute")
async def force_execute_schedule(
schedule_id: int,
background_tasks: BackgroundTasks,
user: User = Depends(current_active_user),
):
"""
Force execution of a specific schedule (for testing/manual triggers).
This bypasses the normal schedule timing and immediately executes
the connector sync for the specified schedule.
"""
try:
scheduler = await get_scheduler()
# Add the force execution to background tasks
background_tasks.add_task(
_force_execute_schedule_task,
scheduler,
schedule_id
)
return {
"message": f"Force execution of schedule {schedule_id} has been queued",
"schedule_id": schedule_id
}
except Exception as e:
logger.error(f"Error force executing schedule {schedule_id}: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to force execute schedule: {e}"
) from e
async def _force_execute_schedule_task(scheduler, schedule_id: int):
"""Background task to force execute a schedule."""
try:
success = await scheduler.force_execute_schedule(schedule_id)
if success:
logger.info(f"Successfully force executed schedule {schedule_id}")
else:
logger.error(f"Failed to force execute schedule {schedule_id}")
except Exception as e:
logger.error(f"Error in force execute task for schedule {schedule_id}: {e}")
@router.get("/scheduler/schedules/upcoming", response_model=list[Dict[str, Any]])
async def get_upcoming_schedules(
limit: int = 10,
session: AsyncSession = Depends(get_async_session),
user: User = Depends(current_active_user),
):
"""
Get a list of upcoming scheduled connector syncs.
Useful for monitoring and debugging scheduled operations.
"""
try:
from app.db import ConnectorSchedule, SearchSourceConnector, SearchSpace
# Get upcoming schedules for connectors owned by the user
query = (
select(ConnectorSchedule)
.options(
selectinload(ConnectorSchedule.connector),
selectinload(ConnectorSchedule.search_space),
)
.join(SearchSourceConnector)
.filter(
SearchSourceConnector.user_id == user.id,
ConnectorSchedule.is_active == True, # noqa: E712
ConnectorSchedule.next_run_at.isnot(None),
)
.order_by(ConnectorSchedule.next_run_at)
.limit(limit)
)
result = await session.execute(query)
schedules = result.scalars().all()
upcoming_schedules = []
for schedule in schedules:
upcoming_schedules.append({
"schedule_id": schedule.id,
"connector_name": schedule.connector.name,
"connector_type": schedule.connector.connector_type.value,
"search_space_name": schedule.search_space.name,
"schedule_type": schedule.schedule_type.value,
"next_run_at": schedule.next_run_at,
"last_run_at": schedule.last_run_at,
"cron_expression": schedule.cron_expression,
})
return upcoming_schedules
except Exception as e:
logger.error(f"Error getting upcoming schedules: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to get upcoming schedules: {e}"
) from e
@router.get("/scheduler/schedules/recent-executions", response_model=list[Dict[str, Any]])
async def get_recent_schedule_executions(
limit: int = 20,
session: AsyncSession = Depends(get_async_session),
user: User = Depends(current_active_user),
):
"""
Get a list of recently executed connector schedules.
Shows the execution history for monitoring and debugging.
"""
try:
from app.db import ConnectorSchedule, SearchSourceConnector
# Get recent executions from the logs table
query = (
select(Log)
.join(SearchSourceConnector, Log.log_metadata["connector_id"].astext.cast(Integer) == SearchSourceConnector.id)
.filter(
SearchSourceConnector.user_id == user.id,
Log.message.like("Scheduled sync%"),
)
.order_by(Log.created_at.desc())
.limit(limit)
)
result = await session.execute(query)
log_rows = result.scalars().all()
executions = []
for log in log_rows:
executions.append({
"log_id": log.id,
"task_name": log.log_metadata.get("task_name") if log.log_metadata else None,
"status": log.status.value,
"level": log.level.value,
"message": log.message,
"source": log.source,
"created_at": log.created_at,
"search_space_id": log.search_space_id,
"connector_id": log.log_metadata.get("connector_id") if log.log_metadata else None,
"schedule_id": log.log_metadata.get("schedule_id") if log.log_metadata else None,
"documents_processed": log.log_metadata.get("documents_processed") if log.log_metadata else None,
})
return executions
except Exception as e:
logger.error(f"Error getting recent schedule executions: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to get recent schedule executions: {e}"
) from e

10
surfsense_backend/app/schemas/__init__.py
View file

@ -8,6 +8,12 @@ from .chats import (
ChatUpdate,
)
from .chunks import ChunkBase, ChunkCreate, ChunkRead, ChunkUpdate
from .connector_schedule import (
ConnectorScheduleBase,
ConnectorScheduleCreate,
ConnectorScheduleRead,
ConnectorScheduleUpdate,
)
from .documents import (
DocumentBase,
DocumentRead,
@ -52,6 +58,10 @@ __all__ = [
"ChunkCreate",
"ChunkRead",
"ChunkUpdate",
"ConnectorScheduleBase",
"ConnectorScheduleCreate",
"ConnectorScheduleRead",
"ConnectorScheduleUpdate",
"DocumentBase",
"DocumentRead",
"DocumentUpdate",

174
surfsense_backend/app/schemas/connector_schedule.py Normal file
View file

@ -0,0 +1,174 @@
from datetime import datetime, time
from typing import Optional
from pydantic import BaseModel, ConfigDict, FieldValidationInfo, field_validator
from app.db import ScheduleType
from .base import IDModel, TimestampModel
class ConnectorScheduleBase(BaseModel):
connector_id: int
search_space_id: int
schedule_type: ScheduleType
cron_expression: str | None = None
is_active: bool = True
# Enhanced time selection options
daily_time: Optional[time] = None # For DAILY schedules (default: 02:00)
weekly_day: Optional[int] = None # For WEEKLY schedules (0=Monday, 6=Sunday, default: 6)
weekly_time: Optional[time] = None # For WEEKLY schedules (default: 02:00)
hourly_minute: Optional[int] = None # For HOURLY schedules (0-59, default: 0)
@field_validator("cron_expression")
@classmethod
def validate_cron_expression(cls, v: str | None, info: FieldValidationInfo) -> str | None:
"""Validate cron expression is provided when schedule_type is CUSTOM."""
schedule_type = info.data.get("schedule_type")
if schedule_type == ScheduleType.CUSTOM and not v:
raise ValueError(
"cron_expression is required when schedule_type is CUSTOM"
)
if schedule_type != ScheduleType.CUSTOM and v:
raise ValueError(
f"cron_expression should only be provided for CUSTOM schedule_type, got {schedule_type}"
)
return v
@field_validator("daily_time")
@classmethod
def validate_daily_time(cls, v: time | None, info: FieldValidationInfo) -> time | None:
"""Validate daily_time is only provided for DAILY schedule type."""
schedule_type = info.data.get("schedule_type")
if v is not None and schedule_type != ScheduleType.DAILY:
raise ValueError(
"daily_time should only be provided for DAILY schedule_type"
)
return v
@field_validator("weekly_day")
@classmethod
def validate_weekly_day(cls, v: int | None, info: FieldValidationInfo) -> int | None:
"""Validate weekly_day is only provided for WEEKLY schedule type."""
schedule_type = info.data.get("schedule_type")
if v is not None and schedule_type != ScheduleType.WEEKLY:
raise ValueError(
"weekly_day should only be provided for WEEKLY schedule_type"
)
if v is not None and not (0 <= v <= 6):
raise ValueError("weekly_day must be between 0 (Monday) and 6 (Sunday)")
return v
@field_validator("weekly_time")
@classmethod
def validate_weekly_time(cls, v: time | None, info: FieldValidationInfo) -> time | None:
"""Validate weekly_time is only provided for WEEKLY schedule type."""
schedule_type = info.data.get("schedule_type")
if v is not None and schedule_type != ScheduleType.WEEKLY:
raise ValueError(
"weekly_time should only be provided for WEEKLY schedule_type"
)
return v
@field_validator("hourly_minute")
@classmethod
def validate_hourly_minute(cls, v: int | None, info: FieldValidationInfo) -> int | None:
"""Validate hourly_minute is only provided for HOURLY schedule type."""
schedule_type = info.data.get("schedule_type")
if v is not None and schedule_type != ScheduleType.HOURLY:
raise ValueError(
"hourly_minute should only be provided for HOURLY schedule_type"
)
if v is not None and not (0 <= v <= 59):
raise ValueError("hourly_minute must be between 0 and 59")
return v
class ConnectorScheduleCreate(ConnectorScheduleBase):
"""Schema for creating a new connector schedule."""
pass
class ConnectorScheduleUpdate(BaseModel):
"""Schema for updating an existing connector schedule."""
schedule_type: ScheduleType | None = None
cron_expression: str | None = None
is_active: bool | None = None
# Enhanced time selection options for updates
daily_time: Optional[time] = None # For DAILY schedules
weekly_day: Optional[int] = None # For WEEKLY schedules (0=Monday, 6=Sunday)
weekly_time: Optional[time] = None # For WEEKLY schedules
hourly_minute: Optional[int] = None # For HOURLY schedules (0-59)
@field_validator("cron_expression")
@classmethod
def validate_cron_expression_update(cls, v: str | None, info: FieldValidationInfo) -> str | None:
"""Validate cron expression for updates."""
schedule_type = info.data.get("schedule_type")
if schedule_type == ScheduleType.CUSTOM and v is None:
raise ValueError(
"cron_expression is required when schedule_type is CUSTOM"
)
return v
@field_validator("daily_time")
@classmethod
def validate_daily_time_update(cls, v: time | None, info: FieldValidationInfo) -> time | None:
"""Validate daily_time is only provided for DAILY schedule type."""
schedule_type = info.data.get("schedule_type")
if v is not None and schedule_type != ScheduleType.DAILY:
raise ValueError(
"daily_time should only be provided for DAILY schedule_type"
)
return v
@field_validator("weekly_day")
@classmethod
def validate_weekly_day_update(cls, v: int | None, info: FieldValidationInfo) -> int | None:
"""Validate weekly_day is only provided for WEEKLY schedule type."""
schedule_type = info.data.get("schedule_type")
if v is not None and schedule_type != ScheduleType.WEEKLY:
raise ValueError(
"weekly_day should only be provided for WEEKLY schedule_type"
)
if v is not None and not (0 <= v <= 6):
raise ValueError("weekly_day must be between 0 (Monday) and 6 (Sunday)")
return v
@field_validator("weekly_time")
@classmethod
def validate_weekly_time_update(cls, v: time | None, info: FieldValidationInfo) -> time | None:
"""Validate weekly_time is only provided for WEEKLY schedule type."""
schedule_type = info.data.get("schedule_type")
if v is not None and schedule_type != ScheduleType.WEEKLY:
raise ValueError(
"weekly_time should only be provided for WEEKLY schedule_type"
)
return v
@field_validator("hourly_minute")
@classmethod
def validate_hourly_minute_update(cls, v: int | None, info: FieldValidationInfo) -> int | None:
"""Validate hourly_minute is only provided for HOURLY schedule type."""
schedule_type = info.data.get("schedule_type")
if v is not None and schedule_type != ScheduleType.HOURLY:
raise ValueError(
"hourly_minute should only be provided for HOURLY schedule_type"
)
if v is not None and not (0 <= v <= 59):
raise ValueError("hourly_minute must be between 0 and 59")
return v
class ConnectorScheduleRead(ConnectorScheduleBase, IDModel, TimestampModel):
"""Schema for reading a connector schedule."""
last_run_at: datetime | None = None
next_run_at: datetime | None = None
model_config = ConfigDict(from_attributes=True)

382
surfsense_backend/app/services/connector_scheduler_service.py Normal file
View file

@ -0,0 +1,382 @@
"""
Connector Scheduler Service
This service manages automated scheduling and execution of connector syncs.
It runs as a background service to check for due schedules and execute them.
"""
import asyncio
import logging
from datetime import datetime, timedelta, timezone
from typing import Dict, List, Optional
from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from app.config import config
from app.db import (
ConnectorSchedule,
SearchSourceConnector,
SearchSourceConnectorType,
get_async_session,
)
from app.services.task_logging_service import TaskLoggingService
from app.tasks.connector_indexers import (
index_airtable_records,
index_clickup_tasks,
index_confluence_pages,
index_discord_messages,
index_github_repos,
index_google_calendar_events,
index_google_gmail_messages,
index_jira_issues,
index_linear_issues,
index_luma_events,
index_notion_pages,
index_slack_messages,
)
from app.utils.schedule_helpers import calculate_next_run
logger = logging.getLogger(__name__)
class ConnectorSchedulerService:
"""Service for managing automated connector scheduling and execution."""
def __init__(self):
self.running = False
self.check_interval = 60 # Check every 60 seconds
self.max_concurrent_jobs = 5 # Maximum concurrent indexing jobs
self.active_jobs: Dict[int, asyncio.Task] = {}
# Mapping of connector types to their indexer functions
self.connector_indexers = {
SearchSourceConnectorType.SLACK_CONNECTOR: index_slack_messages,
SearchSourceConnectorType.NOTION_CONNECTOR: index_notion_pages,
SearchSourceConnectorType.GITHUB_CONNECTOR: index_github_repos,
SearchSourceConnectorType.LINEAR_CONNECTOR: index_linear_issues,
SearchSourceConnectorType.JIRA_CONNECTOR: index_jira_issues,
SearchSourceConnectorType.CONFLUENCE_CONNECTOR: index_confluence_pages,
SearchSourceConnectorType.DISCORD_CONNECTOR: index_discord_messages,
SearchSourceConnectorType.CLICKUP_CONNECTOR: index_clickup_tasks,
SearchSourceConnectorType.GOOGLE_CALENDAR_CONNECTOR: index_google_calendar_events,
SearchSourceConnectorType.GOOGLE_GMAIL_CONNECTOR: index_google_gmail_messages,
SearchSourceConnectorType.AIRTABLE_CONNECTOR: index_airtable_records,
SearchSourceConnectorType.LUMA_CONNECTOR: index_luma_events,
}
async def start(self):
"""Start the scheduler service."""
if self.running:
logger.warning("Scheduler service is already running")
return
self.running = True
logger.info("Starting connector scheduler service...")
try:
while self.running:
await self._check_and_execute_schedules()
await asyncio.sleep(self.check_interval)
except Exception as e:
logger.error(f"Scheduler service error: {e}", exc_info=True)
finally:
self.running = False
logger.info("Connector scheduler service stopped")
async def stop(self):
"""Stop the scheduler service."""
logger.info("Stopping connector scheduler service...")
self.running = False
# Cancel all active jobs
for job_id, task in self.active_jobs.items():
if not task.done():
task.cancel()
logger.info(f"Cancelled active job {job_id}")
self.active_jobs.clear()
async def _check_and_execute_schedules(self):
"""Check for due schedules and execute them."""
try:
async with get_async_session() as session:
# Get all active schedules that are due
due_schedules = await self._get_due_schedules(session)
if not due_schedules:
return
logger.info(f"Found {len(due_schedules)} due schedules")
for schedule in due_schedules:
if len(self.active_jobs) >= self.max_concurrent_jobs:
logger.warning(
f"Maximum concurrent jobs ({self.max_concurrent_jobs}) reached, "
f"skipping schedule {schedule.id}"
)
break
# Execute schedule in background
await self._execute_schedule(schedule, session)
except Exception as e:
logger.error(f"Error checking schedules: {e}", exc_info=True)
async def _get_due_schedules(self, session: AsyncSession) -> List[ConnectorSchedule]:
"""Get all schedules that are due for execution."""
now = datetime.now(timezone.utc)
query = (
select(ConnectorSchedule)
.options(
selectinload(ConnectorSchedule.connector),
selectinload(ConnectorSchedule.search_space),
)
.filter(
ConnectorSchedule.is_active == True, # noqa: E712
ConnectorSchedule.next_run_at <= now,
)
.order_by(ConnectorSchedule.next_run_at)
)
result = await session.execute(query)
return result.scalars().all()
async def _execute_schedule(
self, schedule: ConnectorSchedule, session: AsyncSession
):
"""Execute a scheduled connector sync."""
schedule_id = schedule.id
try:
# Check if we already have an active job for this schedule
if schedule_id in self.active_jobs and not self.active_jobs[schedule_id].done():
logger.warning(f"Schedule {schedule_id} is already running, skipping")
return
# Update last_run_at before starting
await self._update_schedule_last_run(session, schedule_id)
# Get the appropriate indexer function
indexer_func = self.connector_indexers.get(schedule.connector.connector_type)
if not indexer_func:
logger.error(
f"No indexer function found for connector type: {schedule.connector.connector_type}"
)
await self._update_schedule_next_run(session, schedule)
return
# Create a new session for the background task
async with get_async_session() as background_session:
# Start the indexing task
task = asyncio.create_task(
self._run_indexing_task(
background_session,
schedule,
indexer_func,
)
)
self.active_jobs[schedule_id] = task
logger.info(
f"Started scheduled indexing for connector {schedule.connector.name} "
f"(schedule {schedule_id})"
)
except Exception as e:
logger.error(
f"Error executing schedule {schedule_id}: {e}", exc_info=True
)
await self._update_schedule_next_run(session, schedule)
async def _run_indexing_task(
self,
session: AsyncSession,
schedule: ConnectorSchedule,
indexer_func,
):
"""Run the actual indexing task for a schedule."""
schedule_id = schedule.id
connector = schedule.connector
try:
# Create a task log entry
task_logger = TaskLoggingService(session, schedule.search_space_id)
log_entry = await task_logger.log_task_start(
f"scheduled_sync_{connector.connector_type.value}",
"connector_scheduler",
f"Scheduled sync for {connector.name}",
{
"schedule_id": schedule_id,
"connector_id": connector.id,
"connector_type": connector.connector_type.value,
"search_space_id": schedule.search_space_id,
},
)
# Calculate date range for incremental sync
start_date = None
if connector.last_indexed_at:
# Start from last indexed date minus 1 day for overlap
start_date = (connector.last_indexed_at - timedelta(days=1)).strftime(
"%Y-%m-%d"
)
end_date = datetime.now(timezone.utc).strftime("%Y-%m-%d")
# Execute the indexer function
documents_processed, error_message = await indexer_func(
session=session,
connector_id=connector.id,
search_space_id=schedule.search_space_id,
user_id=str(connector.user_id),
start_date=start_date,
end_date=end_date,
update_last_indexed=True,
)
if error_message:
await task_logger.log_task_failure(
log_entry,
f"Scheduled sync failed for {connector.name}",
error_message,
{"documents_processed": documents_processed},
)
logger.error(
f"Scheduled sync failed for connector {connector.name}: {error_message}"
)
else:
await task_logger.log_task_success(
log_entry,
f"Scheduled sync completed for {connector.name}",
{"documents_processed": documents_processed},
)
logger.info(
f"Scheduled sync completed for connector {connector.name}: "
f"{documents_processed} documents processed"
)
# Update next run time
async with get_async_session() as update_session:
await self._update_schedule_next_run(update_session, schedule)
except Exception as e:
logger.error(
f"Error in scheduled indexing task for schedule {schedule_id}: {e}",
exc_info=True,
)
# Update next run time even on error to prevent stuck schedules
async with get_async_session() as update_session:
await self._update_schedule_next_run(update_session, schedule)
finally:
# Remove from active jobs
self.active_jobs.pop(schedule_id, None)
async def _update_schedule_last_run(self, session: AsyncSession, schedule_id: int):
"""Update the last_run_at timestamp for a schedule."""
await session.execute(
update(ConnectorSchedule)
.filter(ConnectorSchedule.id == schedule_id)
.values(last_run_at=datetime.now(timezone.utc))
)
await session.commit()
async def _update_schedule_next_run(
self, session: AsyncSession, schedule: ConnectorSchedule
):
"""Update the next_run_at timestamp for a schedule."""
next_run = calculate_next_run(
schedule.schedule_type,
schedule.cron_expression,
schedule.daily_time,
schedule.weekly_day,
schedule.weekly_time,
schedule.hourly_minute
)
await session.execute(
update(ConnectorSchedule)
.filter(ConnectorSchedule.id == schedule.id)
.values(next_run_at=next_run)
)
await session.commit()
logger.info(
f"Updated next run time for schedule {schedule.id} to {next_run}"
)
async def get_scheduler_status(self) -> dict:
"""Get the current status of the scheduler service."""
return {
"running": self.running,
"active_jobs": len(self.active_jobs),
"max_concurrent_jobs": self.max_concurrent_jobs,
"check_interval": self.check_interval,
"active_job_details": {
str(schedule_id): {
"running": not task.done(),
"done": task.done(),
"cancelled": task.cancelled(),
}
for schedule_id, task in self.active_jobs.items()
},
}
async def force_execute_schedule(self, schedule_id: int) -> bool:
"""Force execution of a specific schedule (for testing/manual triggers)."""
try:
async with get_async_session() as session:
# Get the schedule
result = await session.execute(
select(ConnectorSchedule)
.options(
selectinload(ConnectorSchedule.connector),
selectinload(ConnectorSchedule.search_space),
)
.filter(ConnectorSchedule.id == schedule_id)
)
schedule = result.scalars().first()
if not schedule:
logger.error(f"Schedule {schedule_id} not found")
return False
if not schedule.is_active:
logger.error(f"Schedule {schedule_id} is not active")
return False
# Execute the schedule
await self._execute_schedule(schedule, session)
return True
except Exception as e:
logger.error(f"Error force executing schedule {schedule_id}: {e}")
return False
# Global scheduler instance
_scheduler_instance: Optional[ConnectorSchedulerService] = None
async def get_scheduler() -> ConnectorSchedulerService:
"""Get the global scheduler instance."""
global _scheduler_instance
if _scheduler_instance is None:
_scheduler_instance = ConnectorSchedulerService()
return _scheduler_instance
async def start_scheduler():
"""Start the global scheduler service."""
scheduler = await get_scheduler()
await scheduler.start()
async def stop_scheduler():
"""Stop the global scheduler service."""
global _scheduler_instance
if _scheduler_instance:
await _scheduler_instance.stop()
_scheduler_instance = None

98
surfsense_backend/app/utils/schedule_helpers.py Normal file
View file

@ -0,0 +1,98 @@
"""Helper utilities for calculating schedule next run times."""
from datetime import datetime, timedelta, time, timezone
from croniter import croniter
from app.db import ScheduleType
def calculate_next_run(
schedule_type: ScheduleType,
cron_expression: str | None = None,
daily_time: time | None = None,
weekly_day: int | None = None,
weekly_time: time | None = None,
hourly_minute: int | None = None,
) -> datetime:
"""
Calculate the next run time based on schedule type with enhanced time options.
Args:
schedule_type: The type of schedule (HOURLY, DAILY, WEEKLY, CUSTOM)
cron_expression: Optional cron expression for CUSTOM type
daily_time: Optional time for DAILY schedules (default: 02:00)
weekly_day: Optional day for WEEKLY schedules (0=Monday, 6=Sunday, default: 6)
weekly_time: Optional time for WEEKLY schedules (default: 02:00)
hourly_minute: Optional minute for HOURLY schedules (0-59, default: 0)
Returns:
datetime: The next scheduled run time (timezone-aware in UTC)
Raises:
ValueError: If schedule_type is CUSTOM but no cron_expression provided
"""
now = datetime.now(timezone.utc)
if schedule_type == ScheduleType.HOURLY:
# Run at the specified minute of the next hour
minute = hourly_minute if hourly_minute is not None else 0
next_run = (now + timedelta(hours=1)).replace(minute=minute, second=0, microsecond=0)
return next_run.replace(tzinfo=timezone.utc)
elif schedule_type == ScheduleType.DAILY:
# Run at the specified time next day
target_time = daily_time if daily_time is not None else time(2, 0) # Default 2 AM
next_run = now.replace(hour=target_time.hour, minute=target_time.minute, second=0, microsecond=0)
if next_run <= now:
next_run += timedelta(days=1)
return next_run.replace(tzinfo=timezone.utc)
elif schedule_type == ScheduleType.WEEKLY:
# Run on the specified day at the specified time
target_day = weekly_day if weekly_day is not None else 6 # Default Sunday
target_time = weekly_time if weekly_time is not None else time(2, 0) # Default 2 AM
next_run = now.replace(hour=target_time.hour, minute=target_time.minute, second=0, microsecond=0)
# Calculate days until target day
days_until_target = (target_day - now.weekday()) % 7
if days_until_target == 0 and next_run <= now:
days_until_target = 7
next_run += timedelta(days=days_until_target)
return next_run.replace(tzinfo=timezone.utc)
elif schedule_type == ScheduleType.CUSTOM:
if not cron_expression:
raise ValueError("cron_expression is required for CUSTOM schedule type")
try:
cron = croniter(cron_expression, now)
next_run = cron.get_next(datetime)
# Ensure the returned datetime is timezone-aware in UTC
if next_run.tzinfo is None:
next_run = next_run.replace(tzinfo=timezone.utc)
return next_run
except Exception as e:
raise ValueError(f"Invalid cron expression: {cron_expression}") from e
else:
raise ValueError(f"Unknown schedule type: {schedule_type}")
def is_valid_cron_expression(expression: str) -> bool:
"""
Validate a cron expression.
Args:
expression: The cron expression to validate
Returns:
bool: True if valid, False otherwise
"""
try:
croniter(expression)
return True
except Exception:
return False

572
surfsense_web/app/dashboard/[search_space_id]/connectors/schedules/page.tsx Normal file
View file

@ -0,0 +1,572 @@
"use client";
import { useState, useEffect } from "react";
import { Card, CardContent, CardDescription, CardHeader, CardTitle } from "@/components/ui/card";
import { Button } from "@/components/ui/button";
import { Badge } from "@/components/ui/badge";
import { Tabs, TabsContent, TabsList, TabsTrigger } from "@/components/ui/tabs";
import {
Table,
TableBody,
TableCell,
TableHead,
TableHeader,
TableRow
} from "@/components/ui/table";
import {
Dialog,
DialogContent,
DialogDescription,
DialogHeader,
DialogTitle,
DialogTrigger
} from "@/components/ui/dialog";
import {
Select,
SelectContent,
SelectItem,
SelectTrigger,
SelectValue,
} from "@/components/ui/select";
import { Input } from "@/components/ui/input";
import { Label } from "@/components/ui/label";
import { Switch } from "@/components/ui/switch";
import {
Clock,
Calendar,
Play,
Pause,
Settings,
Plus,
RefreshCw,
CheckCircle,
XCircle,
AlertCircle
} from "lucide-react";
import { toast } from "sonner";
import { format } from "date-fns";
interface Connector {
id: number;
name: string;
connector_type: string;
is_indexable: boolean;
}
interface ConnectorSchedule {
id: number;
connector_id: number;
search_space_id: number;
schedule_type: "HOURLY" | "DAILY" | "WEEKLY" | "CUSTOM";
cron_expression?: string;
is_active: boolean;
last_run_at?: string;
next_run_at?: string;
connector?: Connector;
}
interface SchedulerStatus {
running: boolean;
active_jobs: number;
max_concurrent_jobs: number;
check_interval: number;
}
export default function ConnectorSchedulesPage({
params,
}: {
params: Promise<{ search_space_id: string }>;
}) {
const [searchSpaceId, setSearchSpaceId] = useState<string>("");
const [schedules, setSchedules] = useState<ConnectorSchedule[]>([]);
const [connectors, setConnectors] = useState<Connector[]>([]);
const [schedulerStatus, setSchedulerStatus] = useState<SchedulerStatus | null>(null);
const [loading, setLoading] = useState(true);
const [isCreateDialogOpen, setIsCreateDialogOpen] = useState(false);
const [newSchedule, setNewSchedule] = useState({
connector_id: "",
schedule_type: "DAILY" as const,
cron_expression: "",
daily_time: "02:00",
weekly_day: "6",
weekly_time: "02:00",
hourly_minute: "0",
is_active: true,
});
useEffect(() => {
const resolveParams = async () => {
const resolved = await params;
setSearchSpaceId(resolved.search_space_id);
};
resolveParams();
}, [params]);
useEffect(() => {
if (searchSpaceId) {
fetchSchedules();
fetchConnectors();
fetchSchedulerStatus();
}
}, [searchSpaceId]);
const fetchSchedules = async () => {
try {
const response = await fetch(`/api/v1/connector-schedules/?search_space_id=${searchSpaceId}`);
if (response.ok) {
const data = await response.json();
setSchedules(data);
}
} catch (error) {
console.error("Failed to fetch schedules:", error);
toast.error("Failed to fetch schedules");
}
};
const fetchConnectors = async () => {
try {
const response = await fetch("/api/v1/search-source-connectors/");
if (response.ok) {
const data = await response.json();
// Filter only indexable connectors
setConnectors(data.filter((c: Connector) => c.is_indexable));
}
} catch (error) {
console.error("Failed to fetch connectors:", error);
toast.error("Failed to fetch connectors");
}
};
const fetchSchedulerStatus = async () => {
try {
const response = await fetch("/api/v1/scheduler/status");
if (response.ok) {
const data = await response.json();
setSchedulerStatus(data);
}
} catch (error) {
console.error("Failed to fetch scheduler status:", error);
}
};
const createSchedule = async () => {
try {
const scheduleData = {
connector_id: parseInt(newSchedule.connector_id),
search_space_id: parseInt(searchSpaceId),
schedule_type: newSchedule.schedule_type,
cron_expression: newSchedule.schedule_type === "CUSTOM" ? newSchedule.cron_expression : undefined,
daily_time: newSchedule.schedule_type === "DAILY" ? newSchedule.daily_time : undefined,
weekly_day: newSchedule.schedule_type === "WEEKLY" ? parseInt(newSchedule.weekly_day) : undefined,
weekly_time: newSchedule.schedule_type === "WEEKLY" ? newSchedule.weekly_time : undefined,
hourly_minute: newSchedule.schedule_type === "HOURLY" ? parseInt(newSchedule.hourly_minute) : undefined,
is_active: newSchedule.is_active,
};
const response = await fetch("/api/v1/connector-schedules/", {
method: "POST",
headers: {
"Content-Type": "application/json",
},
body: JSON.stringify(scheduleData),
});
if (response.ok) {
toast.success("Schedule created successfully");
setIsCreateDialogOpen(false);
setNewSchedule({
connector_id: "",
schedule_type: "DAILY",
cron_expression: "",
daily_time: "02:00",
weekly_day: "6",
weekly_time: "02:00",
hourly_minute: "0",
is_active: true,
});
fetchSchedules();
} else {
const error = await response.json();
toast.error(error.detail || "Failed to create schedule");
}
} catch (error) {
console.error("Failed to create schedule:", error);
toast.error("Failed to create schedule");
}
};
const toggleSchedule = async (scheduleId: number, isActive: boolean) => {
try {
const response = await fetch(`/api/v1/connector-schedules/${scheduleId}`, {
method: "PUT",
headers: {
"Content-Type": "application/json",
},
body: JSON.stringify({ is_active: !isActive }),
});
if (response.ok) {
toast.success(`Schedule ${!isActive ? "activated" : "deactivated"}`);
fetchSchedules();
} else {
toast.error("Failed to update schedule");
}
} catch (error) {
console.error("Failed to toggle schedule:", error);
toast.error("Failed to update schedule");
}
};
const forceExecuteSchedule = async (scheduleId: number) => {
try {
const response = await fetch(`/api/v1/scheduler/schedules/${scheduleId}/force-execute`, {
method: "POST",
});
if (response.ok) {
toast.success("Schedule execution started");
fetchSchedules();
} else {
toast.error("Failed to execute schedule");
}
} catch (error) {
console.error("Failed to execute schedule:", error);
toast.error("Failed to execute schedule");
}
};
const getScheduleTypeIcon = (type: string) => {
switch (type) {
case "HOURLY":
return <Clock className="h-4 w-4" />;
case "DAILY":
return <Calendar className="h-4 w-4" />;
case "WEEKLY":
return <Calendar className="h-4 w-4" />;
case "CUSTOM":
return <Settings className="h-4 w-4" />;
default:
return <Clock className="h-4 w-4" />;
}
};
const getScheduleTypeLabel = (type: string) => {
switch (type) {
case "HOURLY":
return "Hourly";
case "DAILY":
return "Daily";
case "WEEKLY":
return "Weekly";
case "CUSTOM":
return "Custom";
default:
return type;
}
};
const getConnectorName = (connectorId: number) => {
const connector = connectors.find(c => c.id === connectorId);
return connector?.name || `Connector ${connectorId}`;
};
if (loading) {
return (
<div className="flex items-center justify-center h-64">
<RefreshCw className="h-8 w-8 animate-spin" />
</div>
);
}
return (
<div className="space-y-6">
<div className="flex items-center justify-between">
<div>
<h1 className="text-3xl font-bold">Connector Schedules</h1>
<p className="text-muted-foreground">
Automate your connector syncs with scheduled indexing
</p>
</div>
<Dialog open={isCreateDialogOpen} onOpenChange={setIsCreateDialogOpen}>
<DialogTrigger asChild>
<Button>
<Plus className="h-4 w-4 mr-2" />
Create Schedule
</Button>
</DialogTrigger>
<DialogContent className="sm:max-w-[500px]">
<DialogHeader>
<DialogTitle>Create New Schedule</DialogTitle>
<DialogDescription>
Set up automated syncing for your connectors
</DialogDescription>
</DialogHeader>
<div className="space-y-4">
<div>
<Label htmlFor="connector">Connector</Label>
<Select
value={newSchedule.connector_id}
onValueChange={(value) => setNewSchedule({ ...newSchedule, connector_id: value })}
>
<SelectTrigger>
<SelectValue placeholder="Select a connector" />
</SelectTrigger>
<SelectContent>
{connectors.map((connector) => (
<SelectItem key={connector.id} value={connector.id.toString()}>
{connector.name} ({connector.connector_type})
</SelectItem>
))}
</SelectContent>
</Select>
</div>
<div>
<Label htmlFor="schedule_type">Schedule Type</Label>
<Select
value={newSchedule.schedule_type}
onValueChange={(value: any) => setNewSchedule({ ...newSchedule, schedule_type: value })}
>
<SelectTrigger>
<SelectValue />
</SelectTrigger>
<SelectContent>
<SelectItem value="HOURLY">Hourly</SelectItem>
<SelectItem value="DAILY">Daily</SelectItem>
<SelectItem value="WEEKLY">Weekly</SelectItem>
<SelectItem value="CUSTOM">Custom (Cron)</SelectItem>
</SelectContent>
</Select>
</div>
{newSchedule.schedule_type === "DAILY" && (
<div>
<Label htmlFor="daily_time">Time</Label>
<Input
type="time"
value={newSchedule.daily_time}
onChange={(e) => setNewSchedule({ ...newSchedule, daily_time: e.target.value })}
/>
</div>
)}
{newSchedule.schedule_type === "WEEKLY" && (
<div className="grid grid-cols-2 gap-4">
<div>
<Label htmlFor="weekly_day">Day of Week</Label>
<Select
value={newSchedule.weekly_day}
onValueChange={(value) => setNewSchedule({ ...newSchedule, weekly_day: value })}
>
<SelectTrigger>
<SelectValue />
</SelectTrigger>
<SelectContent>
<SelectItem value="0">Monday</SelectItem>
<SelectItem value="1">Tuesday</SelectItem>
<SelectItem value="2">Wednesday</SelectItem>
<SelectItem value="3">Thursday</SelectItem>
<SelectItem value="4">Friday</SelectItem>
<SelectItem value="5">Saturday</SelectItem>
<SelectItem value="6">Sunday</SelectItem>
</SelectContent>
</Select>
</div>
<div>
<Label htmlFor="weekly_time">Time</Label>
<Input
type="time"
value={newSchedule.weekly_time}
onChange={(e) => setNewSchedule({ ...newSchedule, weekly_time: e.target.value })}
/>
</div>
</div>
)}
{newSchedule.schedule_type === "HOURLY" && (
<div>
<Label htmlFor="hourly_minute">Minute (0-59)</Label>
<Input
type="number"
min="0"
max="59"
value={newSchedule.hourly_minute}
onChange={(e) => setNewSchedule({ ...newSchedule, hourly_minute: e.target.value })}
/>
</div>
)}
{newSchedule.schedule_type === "CUSTOM" && (
<div>
<Label htmlFor="cron_expression">Cron Expression</Label>
<Input
placeholder="0 2 * * *"
value={newSchedule.cron_expression}
onChange={(e) => setNewSchedule({ ...newSchedule, cron_expression: e.target.value })}
/>
</div>
)}
<div className="flex items-center space-x-2">
<Switch
id="is_active"
checked={newSchedule.is_active}
onCheckedChange={(checked) => setNewSchedule({ ...newSchedule, is_active: checked })}
/>
<Label htmlFor="is_active">Active</Label>
</div>
<div className="flex justify-end space-x-2">
<Button variant="outline" onClick={() => setIsCreateDialogOpen(false)}>
Cancel
</Button>
<Button onClick={createSchedule}>
Create Schedule
</Button>
</div>
</div>
</DialogContent>
</Dialog>
</div>
{/* Scheduler Status */}
{schedulerStatus && (
<Card>
<CardHeader>
<CardTitle className="flex items-center gap-2">
<Settings className="h-5 w-5" />
Scheduler Status
</CardTitle>
</CardHeader>
<CardContent>
<div className="grid grid-cols-1 md:grid-cols-4 gap-4">
<div className="text-center">
<div className="text-2xl font-bold">
{schedulerStatus.running ? (
<CheckCircle className="h-8 w-8 text-green-500 mx-auto" />
) : (
<XCircle className="h-8 w-8 text-red-500 mx-auto" />
)}
</div>
<p className="text-sm text-muted-foreground">Status</p>
</div>
<div className="text-center">
<div className="text-2xl font-bold">{schedulerStatus.active_jobs}</div>
<p className="text-sm text-muted-foreground">Active Jobs</p>
</div>
<div className="text-center">
<div className="text-2xl font-bold">{schedulerStatus.max_concurrent_jobs}</div>
<p className="text-sm text-muted-foreground">Max Concurrent</p>
</div>
<div className="text-center">
<div className="text-2xl font-bold">{schedulerStatus.check_interval}s</div>
<p className="text-sm text-muted-foreground">Check Interval</p>
</div>
</div>
</CardContent>
</Card>
)}
{/* Schedules Table */}
<Card>
<CardHeader>
<CardTitle>Scheduled Connectors</CardTitle>
<CardDescription>
Manage automated sync schedules for your connectors
</CardDescription>
</CardHeader>
<CardContent>
{schedules.length === 0 ? (
<div className="text-center py-8">
<Calendar className="h-12 w-12 text-muted-foreground mx-auto mb-4" />
<h3 className="text-lg font-semibold">No schedules configured</h3>
<p className="text-muted-foreground mb-4">
Create your first schedule to automate connector syncing
</p>
<Button onClick={() => setIsCreateDialogOpen(true)}>
<Plus className="h-4 w-4 mr-2" />
Create Schedule
</Button>
</div>
) : (
<Table>
<TableHeader>
<TableRow>
<TableHead>Connector</TableHead>
<TableHead>Schedule</TableHead>
<TableHead>Status</TableHead>
<TableHead>Last Run</TableHead>
<TableHead>Next Run</TableHead>
<TableHead>Actions</TableHead>
</TableRow>
</TableHeader>
<TableBody>
{schedules.map((schedule) => (
<TableRow key={schedule.id}>
<TableCell className="font-medium">
{getConnectorName(schedule.connector_id)}
</TableCell>
<TableCell>
<div className="flex items-center gap-2">
{getScheduleTypeIcon(schedule.schedule_type)}
<span>{getScheduleTypeLabel(schedule.schedule_type)}</span>
{schedule.cron_expression && (
<Badge variant="outline" className="text-xs">
{schedule.cron_expression}
</Badge>
)}
</div>
</TableCell>
<TableCell>
<Badge variant={schedule.is_active ? "default" : "secondary"}>
{schedule.is_active ? "Active" : "Inactive"}
</Badge>
</TableCell>
<TableCell>
{schedule.last_run_at ? (
<span className="text-sm">
{format(new Date(schedule.last_run_at), "MMM d, HH:mm")}
</span>
) : (
<span className="text-muted-foreground">Never</span>
)}
</TableCell>
<TableCell>
{schedule.next_run_at ? (
<span className="text-sm">
{format(new Date(schedule.next_run_at), "MMM d, HH:mm")}
</span>
) : (
<span className="text-muted-foreground">-</span>
)}
</TableCell>
<TableCell>
<div className="flex items-center gap-2">
<Button
size="sm"
variant="outline"
onClick={() => toggleSchedule(schedule.id, schedule.is_active)}
>
{schedule.is_active ? <Pause className="h-4 w-4" /> : <Play className="h-4 w-4" />}
</Button>
<Button
size="sm"
variant="outline"
onClick={() => forceExecuteSchedule(schedule.id)}
disabled={!schedule.is_active}
>
<RefreshCw className="h-4 w-4" />
</Button>
</div>
</TableCell>
</TableRow>
))}
</TableBody>
</Table>
)}
</CardContent>
</Card>
</div>
);
}

28863
surfsense_web/package-lock.json generated
View file

File diff suppressed because it is too large Load diff