feat: add schedule-owned logical parameter bindings

README.md

@@ -61,9 +61,9 @@ Define connection ─▶ Author SQL (with :bind params) ─▶ Attach auth ─
 ```
 
 1. **Connect** to your Oracle database with securely stored, encrypted credentials.
-2. **Author** a `SELECT` query using named bind parameters (`:param_name`) in a rich SQL editor. The wizard detects parameters as you type, requests temporary sample values before previewing, and supports fixed or dynamic date defaults (`today` and `yesterday`). Snapshot endpoints require a default for every bind so scheduled refreshes can execute without request inputs. Scheduler defaults do not make required parameters optional for live HTTP requests; callers must supply required dates as either `YYYY-MM-DD` or `DD-MM-YYYY`.
+2. **Author** a `SELECT` query using named bind parameters (`:param_name`) in a rich SQL editor. The wizard detects parameters as you type, requests temporary sample values before previewing, and supports optional endpoint defaults for live requests. Required live parameters remain mandatory; callers can supply dates as either `YYYY-MM-DD` or `DD-MM-YYYY`.
 3. **Secure** the endpoint by attaching a Bearer token, Basic Auth, or API key policy.
-4. **Choose** a data strategy: serve results **live** on each request, or from a **scheduled snapshot** cache.
+4. **Choose** a data strategy: serve results **live** on each request, or from a **scheduled snapshot** cache. Snapshot schedules own their parameter values: bind a fixed value, explicit SQL `NULL`, logical run date, relative date, or a reusable calendar window, then preview the next three resolved runs before creating the schedule.
 5. **Publish** a versioned endpoint under `/api/v1/data/*` that resolves dynamically — no service restart needed.
 
 ## Features
@@ -75,7 +75,7 @@ QueryGateway is organized into five admin modules, all driven from the React adm
 | **Connections** | Create, edit, test, and delete Oracle database connections. Credentials are encrypted at rest; pool sizing and timeouts are configurable. Uses `python-oracledb`; thin mode needs no native client, while the backend Docker image includes Oracle Instant Client 19.32 for thick mode. |
 | **API Creation Wizard** | A multi-step wizard that turns a parameterized SQL query into a deployable GET endpoint: pick a connection, author SQL with a rich editor, supply preview-only sample values for detected bind parameters, configure fixed values, explicit SQL `NULL` values for optional binds, or dynamic date defaults, preview sample rows and inferred schema, map/rename output columns, attach an auth method, and select a data strategy. |
 | **Authentication** | Manage per-endpoint auth methods — Bearer token (JWT), Basic Auth, and API key. Tokens are issued/verified with `PyJWT`; credentials are hashed with `bcrypt`. Every `/api/v1/data/*` request is authenticated; endpoints without a dedicated method require the platform admin Bearer token. |
-| **Scheduling & Snapshots** | Schedule query refreshes with friendly hourly, daily, weekly, or monthly calendar controls, an advanced custom-cron option, or a fixed interval. Run now, pause/resume, and enable/disable jobs. Results are cached as PostgreSQL JSONB snapshots and served with freshness metadata. Schedule definitions are persisted in the app database, active APScheduler jobs are restored on API startup, and deleting a schedule preserves its job-run and snapshot history. Deleting an endpoint removes its schedule and snapshots while retaining orphaned job-run audit records. |
+| **Scheduling & Snapshots** | Schedule query refreshes with friendly hourly, daily, weekly, or monthly calendar controls, an advanced custom-cron option, or a fixed interval. Each schedule has an IANA timezone and explicit per-parameter sources: fixed value, SQL `NULL`, logical run date, relative date, or inclusive calendar-window boundaries such as previous day, last N complete days, week/month to date, previous week, and previous month. Preview the next three resolved runs, run now, pause/resume, and inspect persisted logical-date, window, and resolved-bind audit context. Results are cached as PostgreSQL JSONB snapshots. Active APScheduler jobs are restored on API startup; deletion preserves job-run history. |
 | **Settings & Health** | Configure runtime settings (base URL/port, logging level, query timeouts, CORS/rate-limit inputs) and view a health dashboard covering API, PostgreSQL, Oracle connectivity, scheduler status, and recent job outcomes. |
 
 ### Security by Default

backend/alembic/versions/e4a6c2d9f801_add_schedule_parameter_bindings.py

@@ -0,0 +1,132 @@
+"""Add schedule-owned parameter bindings and logical run audit context.
+
+Revision ID: e4a6c2d9f801
+Revises: c7e91a4f2d60
+Create Date: 2026-08-30
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+revision: str = "e4a6c2d9f801"
+down_revision: str | None = "c7e91a4f2d60"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "schedules",
+        sa.Column("timezone", sa.String(length=64), server_default="UTC", nullable=False),
+    )
+    op.add_column(
+        "schedules",
+        sa.Column(
+            "parameter_bindings_json",
+            postgresql.JSONB(astext_type=sa.Text()),
+            server_default=sa.text("'{}'::jsonb"),
+            nullable=False,
+        ),
+    )
+    op.add_column(
+        "schedules",
+        sa.Column(
+            "window_config_json",
+            postgresql.JSONB(astext_type=sa.Text()),
+            nullable=True,
+        ),
+    )
+
+    # Preserve the behavior of existing schedules by copying endpoint defaults
+    # into explicit schedule-owned bindings. Any legacy parameter without a
+    # resolvable default remains absent and will be reported when the schedule
+    # is edited, previewed, or executed rather than silently inventing a value.
+    op.execute(
+        """
+        UPDATE schedules AS schedule
+        SET parameter_bindings_json = COALESCE(migrated.bindings, '{}'::jsonb)
+        FROM (
+            SELECT
+                endpoint.id AS endpoint_id,
+                jsonb_object_agg(
+                    parameter.name,
+                    CASE
+                        WHEN parameter.descriptor->>'default_expression' = 'today'
+                            THEN jsonb_build_object('source', 'run_date')
+                        WHEN parameter.descriptor->>'default_expression' = 'yesterday'
+                            THEN jsonb_build_object(
+                                'source', 'relative_date', 'offset_days', -1
+                            )
+                        WHEN COALESCE(
+                            (parameter.descriptor->>'default_is_null')::boolean,
+                            false
+                        )
+                            THEN jsonb_build_object('source', 'null')
+                        WHEN parameter.descriptor ? 'default'
+                             AND parameter.descriptor->'default' <> 'null'::jsonb
+                            THEN jsonb_build_object(
+                                'source', 'literal',
+                                'value', parameter.descriptor->'default'
+                            )
+                        ELSE NULL
+                    END
+                ) FILTER (
+                    WHERE parameter.descriptor->>'default_expression' IN ('today', 'yesterday')
+                       OR COALESCE(
+                            (parameter.descriptor->>'default_is_null')::boolean,
+                            false
+                       )
+                       OR (
+                            parameter.descriptor ? 'default'
+                            AND parameter.descriptor->'default' <> 'null'::jsonb
+                       )
+                ) AS bindings
+            FROM endpoints AS endpoint
+            CROSS JOIN LATERAL jsonb_each(
+                COALESCE(endpoint.param_schema_json, '{}'::jsonb)
+            ) AS parameter(name, descriptor)
+            GROUP BY endpoint.id
+        ) AS migrated
+        WHERE schedule.endpoint_id = migrated.endpoint_id
+        """
+    )
+
+    op.add_column(
+        "job_runs",
+        sa.Column("scheduled_for", sa.DateTime(timezone=True), nullable=True),
+    )
+    op.add_column("job_runs", sa.Column("logical_date", sa.Date(), nullable=True))
+    op.add_column("job_runs", sa.Column("window_start", sa.Date(), nullable=True))
+    op.add_column("job_runs", sa.Column("window_end", sa.Date(), nullable=True))
+    op.add_column(
+        "job_runs",
+        sa.Column(
+            "resolved_params_json",
+            postgresql.JSONB(astext_type=sa.Text()),
+            nullable=True,
+        ),
+    )
+    op.add_column("job_runs", sa.Column("trigger_source", sa.String(length=20), nullable=True))
+    op.add_column("job_runs", sa.Column("binding_hash", sa.String(length=64), nullable=True))
+    op.create_unique_constraint(
+        "uq_job_runs_schedule_scheduled_for",
+        "job_runs",
+        ["schedule_id", "scheduled_for"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_constraint("uq_job_runs_schedule_scheduled_for", "job_runs", type_="unique")
+    op.drop_column("job_runs", "binding_hash")
+    op.drop_column("job_runs", "trigger_source")
+    op.drop_column("job_runs", "resolved_params_json")
+    op.drop_column("job_runs", "window_end")
+    op.drop_column("job_runs", "window_start")
+    op.drop_column("job_runs", "logical_date")
+    op.drop_column("job_runs", "scheduled_for")
+    op.drop_column("schedules", "window_config_json")
+    op.drop_column("schedules", "parameter_bindings_json")
+    op.drop_column("schedules", "timezone")

backend/app/models/job_run.py

@@ -1,11 +1,12 @@
 """JobRun model — immutable execution audit record for each scheduler invocation."""
 
 import uuid
-from datetime import datetime
+from datetime import date, datetime
 from enum import StrEnum
 
-from sqlalchemy import DateTime, ForeignKey, Integer, String, func
+from sqlalchemy import Date, DateTime, ForeignKey, Integer, String, UniqueConstraint, func
 from sqlalchemy import Enum as SAEnum
+from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.models.base import Base, UUIDPrimaryKeyMixin
@@ -27,6 +28,13 @@ class JobRun(UUIDPrimaryKeyMixin, Base):
     """
 
     __tablename__ = "job_runs"
+    __table_args__ = (
+        UniqueConstraint(
+            "schedule_id",
+            "scheduled_for",
+            name="uq_job_runs_schedule_scheduled_for",
+        ),
+    )
 
     id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
 
@@ -47,6 +55,13 @@ class JobRun(UUIDPrimaryKeyMixin, Base):
     row_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
     # Truncated error detail; full stack trace goes to structured log.
     error_detail: Mapped[str | None] = mapped_column(String(5000), nullable=True)
+    scheduled_for: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    logical_date: Mapped[date | None] = mapped_column(Date, nullable=True)
+    window_start: Mapped[date | None] = mapped_column(Date, nullable=True)
+    window_end: Mapped[date | None] = mapped_column(Date, nullable=True)
+    resolved_params_json: Mapped[dict[str, object] | None] = mapped_column(JSONB, nullable=True)
+    trigger_source: Mapped[str | None] = mapped_column(String(20), nullable=True)
+    binding_hash: Mapped[str | None] = mapped_column(String(64), nullable=True)
 
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), server_default=func.now(), nullable=False

backend/app/models/schedule.py

@@ -6,6 +6,7 @@
 
 from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, text
 from sqlalchemy import Enum as SAEnum
+from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.models.base import Base, TimestampMixin, UUIDPrimaryKeyMixin
@@ -38,13 +39,19 @@ class Schedule(UUIDPrimaryKeyMixin, TimestampMixin, Base):
     cron_expression: Mapped[str | None] = mapped_column(String(100), nullable=True)
     # Used when schedule_type == 'interval'. Positive integer seconds.
     interval_seconds: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    timezone: Mapped[str] = mapped_column(
+        String(64), nullable=False, default="UTC", server_default="UTC"
+    )
+    parameter_bindings_json: Mapped[dict[str, object]] = mapped_column(
+        JSONB,
+        nullable=False,
+        default=dict,
+        server_default=text("'{}'::jsonb"),
+    )
+    window_config_json: Mapped[dict[str, object] | None] = mapped_column(JSONB, nullable=True)
 
     is_active: Mapped[bool] = mapped_column(
         Boolean, default=True, nullable=False, server_default=text("true")
     )
-    last_run_at: Mapped[datetime | None] = mapped_column(
-        DateTime(timezone=True), nullable=True
-    )
-    next_run_at: Mapped[datetime | None] = mapped_column(
-        DateTime(timezone=True), nullable=True
-    )
+    last_run_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    next_run_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)

backend/app/repositories/job_run.py

@@ -2,6 +2,7 @@
 
 import uuid
 from collections.abc import Sequence
+from datetime import datetime
 
 from sqlalchemy import select
 
@@ -20,6 +21,17 @@ class JobRunRepository(BaseCrudRepository[JobRun]):
 
     model = JobRun
 
+    async def get_by_schedule_and_scheduled_for(
+        self, schedule_id: uuid.UUID, scheduled_for: datetime
+    ) -> JobRun | None:
+        result = await self._db.execute(
+            select(JobRun).where(
+                JobRun.schedule_id == schedule_id,
+                JobRun.scheduled_for == scheduled_for,
+            )
+        )
+        return result.scalar_one_or_none()
+
     async def get_all(
         self,
         *,

backend/app/routers/endpoints.py

@@ -32,6 +32,7 @@
     SqlPreviewResponse,
 )
 from app.services.endpoint import EndpointService
+from app.services.schedule_bindings import ScheduleBindingError
 from app.services.scheduler import remove_schedule_job
 
 log = structlog.get_logger()
@@ -47,6 +48,7 @@ def _service(db: AsyncSession = Depends(get_db)) -> EndpointService:
     return EndpointService(
         EndpointRepository(db),
         ConnectionRepository(db),
+        ScheduleRepository(db),
     )
 
 
@@ -113,7 +115,7 @@ async def update_endpoint(
 ) -> EndpointResponse:
     try:
         result = await svc.update_endpoint(endpoint_id, payload)
-    except (PublicEndpointError, SnapshotConfigurationError) as exc:
+    except (PublicEndpointError, ScheduleBindingError, SnapshotConfigurationError) as exc:
         raise HTTPException(
             status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc)
         ) from exc