Epic: Add Workflow Scheduler to Conductor OSS

## ✅ Implementation Complete — PR #782

Implemented and verified. See [PR #782](https://github.com/conductor-oss/conductor/pull/782) for full details.

**What was built:**
- `conductor-scheduler` Gradle module with PostgreSQL-backed scheduling
- Full REST API at `/api/scheduler` (create/read/update/delete, pause/resume, history, next-execution preview)
- Spring 6-field cron with timezone/DST support, schedule start/end bounds, catchup mode
- `scheduledTime` + `executionTime` injected into every triggered workflow's input (matches Orkes behavior)
- Configurable dispatch jitter (`jitter-max-ms`) to spread thundering-herd load across a small time window
- Execution history with auto-pruning, stale-POLLED cleanup
- 83 tests passing (unit + Testcontainers integration)
- 8 example workflows verified live; concurrency tests #9–12 verified with two machines

**What was deferred (not in this PR):**
- Redis caching layer (`SchedulerRedisCacheDAO`) — out of scope for OSS initial port, can be added as a follow-on
- Search/filter API beyond `?workflowName=` — future enhancement
- Tags, bulk operations, UI integration — future enhancement

---

## Updated Success Criteria

- ✅ Schedules execute workflows at correct times (within 1 second accuracy)
- ✅ Timezone/DST handling works correctly
- ✅ Pause/resume, bounds, catchup mode work as in Orkes
- ✅ Execution history tracked and cleaned up (configurable retention)
- ⏭️ Optional Redis caching — deferred to future PR
- ✅ No enterprise dependencies visible to users (RBAC, tags)
- ✅ Test coverage > 80% (83 tests across 5 test classes)
- ✅ **Convergence requirement: DAO interface matches Orkes exactly**
- ✅ **Orkes can adopt OSS scheduler module with minimal changes (orgId injection)**

---

## Overview

This epic ports the proven workflow scheduler implementation from Orkes Conductor to OSS, removing enterprise-specific dependencies (RBAC, tags, search) while **keeping the same interface and data model** to enable future convergence where Orkes can adopt the OSS scheduler directly.

## Problem Statement

Conductor OSS does not have workflow scheduling capabilities. Users must use external tools (cron, Airflow, Kubernetes CronJobs) or build custom solutions. Orkes Conductor has a production-proven scheduler that can be ported to OSS.

## Convergence Strategy

**Key Principle**: OSS scheduler must use the **same interface** as Orkes scheduler so Orkes can potentially adopt OSS implementation directly.

- **Same DAO interface** - Method signatures match Orkes exactly
- **Same data model** - `orgId` field exists (OSS always uses `"default"`)
- **Same database schema** - Compatible with Orkes (composite key on `org_id, schedule_name`)
- **Different implementations** - OSS ignores/defaults orgId, Orkes uses it for multi-tenancy

This allows Orkes to either:
1. Use OSS scheduler module directly, OR
2. Extend OSS with minimal changes for enterprise features

## Orkes vs OSS: What Changes

| Component | Orkes Implementation | OSS Implementation | Change Type |
|-----------|---------------------|-------------------|-------------|
| **Core Scheduling** |
| Cron parsing | Spring `CronExpression` (6-field, second precision) | ✅ Port directly | Keep |
| Next-run calculation | `ZonedDateTime` with timezone/DST support | ✅ Port directly | Keep |
| Queue-based execution | QueueDAO + delayed execution | ✅ Port directly | Keep |
| Workflow triggering | WorkflowService integration | ✅ Port directly (already in OSS) | Keep |
| **Data Model** |
| `orgId` field | Required, from request context | ✅ **Keep field, always use `"default"`** | **Simplify** |
| Cron expression | 6-field format string | ✅ Same | Keep |
| Timezone | `zoneId` field (String) | ✅ Same | Keep |
| Pause/resume | Boolean flag + reason | ✅ Same | Keep |
| Schedule bounds | `scheduleStartTime`/`scheduleEndTime` | ✅ Same | Keep |
| Catchup mode | `runCatchupScheduleInstances` boolean | ✅ Same | Keep |
| Tags | `List` field | ❌ Remove from model | Remove |
| User tracking | `createdBy`/`updatedBy` (enforced) | Make optional, no enforcement | Simplify |
| **DAO Interface** |
| Method signatures | `findScheduleByName(String orgId, String name)` | ✅ **Same signature, OSS passes `"default"`** | **Simplify** |
| Permission methods | `getAllSchedulesWithPermissions(...)` | ❌ Omit permission variants | Remove |
| Search methods | `searchSchedules(...)` | ❌ Omit (can add later) | Remove |
| **Persistence** |
| PostgreSQL schema | `PRIMARY KEY (org_id, schedule_name)` | ✅ **Same schema, org_id defaults to `"default"`** | **Simplify** |
| PostgreSQL DAO | `PostgresSchedulerDAO` with orgId queries | ✅ Port directly, use `"default"` orgId | Simplify |
| Redis caching | `SchedulerRedisCacheDAO` (optional) | ✅ Port directly, use `"default"` orgId | Simplify |
| **Execution History** |
| Tracking states | POLLED → EXECUTED/FAILED | ✅ Same | Keep |
| Retention policy | Keep last 5 records + time-based cleanup | ✅ Same (configurable via properties) | Keep |
| Archival tables | Separate archival tables + maintenance | Simple cleanup job on main table | Simplify |
| **REST API** |
| Endpoint paths | `/api/scheduler/schedules/{name}` | ✅ Same (orgId injected as `"default"` server-side) | Simplify |
| orgId parameter | Explicit in Orkes API | Hidden from users (always `"default"`) | Simplify |
| Bulk operations | Create/update/delete multiple schedules | ❌ Omit (can add later) | Remove |
| **Security** |
| Permission checks | RBAC (`subjects`, `access` parameters) | ❌ Remove all permission checks | Remove |
| Secure DAO wrapper | `SecurePostgresSchedulerArchiveDAO` | Use base DAO without wrapper | Remove |
| **Configuration** |
| Properties | `SchedulerProperties` class | ✅ Port directly (same defaults) | Keep |
| Feature flags | `conductor.scheduler.enabled` | ✅ Same | Keep |

## Architecture (Same Interface as Orkes)

**REST API (orgId hidden from OSS users):**
```
POST /api/scheduler/schedules
GET /api/scheduler/schedules/{name}
GET /api/scheduler/schedules
DELETE /api/scheduler/schedules/{name}
PUT /api/scheduler/schedules/{name}/pause
PUT /api/scheduler/schedules/{name}/resume
GET /api/scheduler/schedules/{name}/next-execution-times
```

**Data Model (same as Orkes, minus tags):**
```java
public class WorkflowSchedule {
private String orgId; // Always "default" in OSS
private String name;
private String cronExpression; // 6-field format: "0 0 9 * * MON"
private String zoneId; // e.g., "America/New_York"
private boolean paused;
private String pausedReason;
private StartWorkflowRequest startWorkflowRequest;
private Long scheduleStartTime; // Optional bounds
private Long scheduleEndTime;
private boolean runCatchupScheduleInstances;
private Long createTime;
private Long updatedTime;
private String createdBy; // Optional (no enforcement)
private String updatedBy; // Optional (no enforcement)
private String description;
private Long nextRunTime; // Cached
}
```

**DAO Interface (same signatures as Orkes):**
```java
public interface SchedulerDAO {
// Write operations - orgId from model
void updateSchedule(WorkflowSchedule schedule);
void saveExecutionRecord(WorkflowScheduleExecution execution);

// Read operations - orgId as parameter (OSS always passes "default")
WorkflowSchedule findScheduleByName(String orgId, String name);
List getAllSchedules(String orgId);
List findAllSchedules(String orgId, String workflowName);
void deleteWorkflowSchedule(String orgId, String name);

// Execution tracking
WorkflowScheduleExecution readExecutionRecord(String orgId, String executionId);
void removeExecutionRecord(String orgId, String executionId);
List getPendingExecutionRecordIds(String orgId);

// Next run time management
long getNextRunTimeInEpoch(String orgId, String scheduleName);
void setNextRunTimeInEpoch(String orgId, String name, long epochMillis);
}
```

**Database Schema (compatible with Orkes):**
```sql
CREATE TABLE workflow_schedule (
org_id VARCHAR(255) NOT NULL DEFAULT 'default', -- Always 'default' in OSS
schedule_name VARCHAR(255) NOT NULL,
workflow_name VARCHAR(255) NOT NULL,
json_data TEXT NOT NULL,
next_run_time BIGINT,
PRIMARY KEY (org_id, schedule_name), -- Composite key (same as Orkes)
INDEX workflow_name_idx (workflow_name),
INDEX next_run_time_idx (next_run_time)
);

CREATE TABLE workflow_schedule_execution (
org_id VARCHAR(255) NOT NULL DEFAULT 'default', -- Always 'default' in OSS
execution_id VARCHAR(255) NOT NULL,
schedule_name VARCHAR(255) NOT NULL,
workflow_id VARCHAR(255),
scheduled_time BIGINT NOT NULL,
execution_time BIGINT NOT NULL,
state VARCHAR(50) NOT NULL, -- POLLED, EXECUTED, FAILED
reason TEXT,
zone_id VARCHAR(50),
PRIMARY KEY (org_id, execution_id), -- Composite key (same as Orkes)
INDEX schedule_name_idx (schedule_name),
INDEX execution_time_idx (execution_time)
);
```

## Implementation Plan

### Phase 1: Core Module Setup
- [ ] Create `conductor-scheduler` module
- [ ] Port `WorkflowSchedule` from Orkes (keep orgId field, remove tags)
- [ ] Port `WorkflowScheduleExecution` from Orkes (keep orgId field)
- [ ] Port `SchedulerDAO` interface (keep orgId parameters, remove permission methods)
- [ ] Port `PostgresSchedulerDAO` (keep orgId in queries, OSS uses "default")
- [ ] Port `SchedulerRedisCacheDAO` (optional, keep orgId handling)
- [ ] Add PostgreSQL migration scripts with orgId DEFAULT 'default'
- [ ] Port `SchedulerProperties` configuration
- [ ] Add constant: `public static final String DEFAULT_ORG_ID = "default";`

**Module structure:**
```
conductor-scheduler/
└── src/main/java/com/netflix/conductor/scheduler/
├── model/
│ ├── WorkflowSchedule.java
│ └── WorkflowScheduleExecution.java
├── dao/
│ ├── SchedulerDAO.java
│ ├── postgres/PostgresSchedulerDAO.java
│ └── redis/SchedulerRedisCacheDAO.java (optional)
├── config/
│ ├── SchedulerConfiguration.java
│ ├── SchedulerProperties.java
│ └── SchedulerConditions.java
└── service/
├── SchedulerService.java
└── SchedulerTimeProvider.java
```

### Phase 2: Scheduler Service
- [ ] Port `SchedulerService` (remove permission checks, inject "default" orgId)
- [ ] Port cron parsing, next-run calculation, timezone logic
- [ ] Port queue-based polling and workflow execution triggering
- [ ] Port execution tracking and state management
- [ ] Port pause/resume, schedule bounds, catchup mode
- [ ] Port execution history cleanup job (simplified)
- [ ] Add helper method: `getOrgId()` that returns `DEFAULT_ORG_ID`

**Key implementation detail:**
```java
public class SchedulerService {
private static final String DEFAULT_ORG_ID = "default";

public WorkflowSchedule getSchedule(String name) {
// OSS always uses default orgId
return schedulerDAO.findScheduleByName(DEFAULT_ORG_ID, name);
}

public void createOrUpdateSchedule(WorkflowSchedule schedule) {
// Ensure orgId is set to default
if (schedule.getOrgId() == null) {
schedule.setOrgId(DEFAULT_ORG_ID);
}
schedulerDAO.updateSchedule(schedule);
}
}
```

### Phase 3: REST API & Integration
- [ ] Port `SchedulerResource` REST controller (remove permission checks, inject "default" orgId)
- [ ] Port OpenAPI/Swagger documentation
- [ ] Port validation and error handling
- [ ] Integrate into `conductor-server`
- [ ] Update server build.gradle
- [ ] Ensure REST API never exposes orgId to users (internal only)

**REST API implementation:**
```java
@RestController
@RequestMapping("/api/scheduler")
public class SchedulerResource {
// Users don't see orgId - it's injected as "default" internally

@PostMapping("/schedules")
public WorkflowSchedule createSchedule(@RequestBody WorkflowSchedule schedule) {
schedule.setOrgId(SchedulerService.DEFAULT_ORG_ID); // Inject default
return schedulerService.createOrUpdateSchedule(schedule);
}

@GetMapping("/schedules/{name}")
public WorkflowSchedule getSchedule(@PathVariable String name) {
return schedulerService.getSchedule(name); // Service handles orgId
}
}
```

### Phase 4: Testing & Documentation
- [ ] Port unit tests from Orkes (use "default" orgId)
- [ ] Port integration tests (PostgreSQL testcontainer)
- [ ] Port timezone, pause/resume, catchup tests
- [ ] Port REST API tests
- [ ] Write user documentation (no mention of orgId - internal detail)
- [ ] Document convergence strategy for Orkes team
- [ ] Update CHANGELOG

## Configuration (Same as Orkes)

```properties
conductor.scheduler.enabled=true
conductor.scheduler.polling-thread-count=1
conductor.scheduler.polling-interval=100
conductor.scheduler.poll-batch-size=5
conductor.scheduler.scheduler-time-zone=UTC
conductor.scheduler.archival-max-records=5 # Keep last 5 per schedule
conductor.scheduler.archival-max-record-threshold=10
conductor.scheduler.user-cache-expire-after-write-seconds=120 # Redis cache TTL
```

## User Experience (orgId Hidden)

**Create a schedule (orgId not visible to users):**
```bash
curl -X POST http://localhost:8080/api/scheduler/schedules \
-H "Content-Type: application/json" \
-d '{
"name": "daily-report",
"cronExpression": "0 0 9 * * MON-FRI",
"zoneId": "UTC",
"startWorkflowRequest": {
"name": "report-workflow",
"version": 1
}
}'
```

Note: `orgId` is set to `"default"` automatically by the server. Users never see or set this field.

**Common cron patterns (6-field format):**
```
Every 30 seconds: */30 * * * * *
Every 5 minutes: 0 */5 * * * *
Every hour: 0 0 * * * *
Daily at 9am: 0 0 9 * * *
Weekdays at 9am: 0 0 9 * * MON-FRI
```

## Success Criteria

- ✅ Schedules execute workflows at correct times (within 1 second accuracy)
- ✅ Timezone/DST handling works correctly
- ✅ Pause/resume, bounds, catchup mode work as in Orkes
- ✅ Execution history tracked and cleaned up (5 records + time-based retention)
- ✅ Optional Redis caching works when enabled
- ✅ No enterprise dependencies visible to users (RBAC, tags)
- ✅ Test coverage > 80%
- ✅ **Convergence requirement: DAO interface matches Orkes exactly**
- ✅ **Orkes can potentially adopt OSS scheduler module with minimal changes**

## Convergence Benefits

By keeping the same interface as Orkes:

**For OSS:**
- Proven, production-tested design
- Clear upgrade path if multi-tenancy ever needed
- Minimal complexity (orgId always "default")

**For Orkes:**
- Can potentially drop custom scheduler implementation
- Use OSS as shared core
- Add multi-tenancy by injecting actual orgId from `OrkesRequestContext`
- Reduced maintenance burden

**Migration path for Orkes:**
1. OSS scheduler reaches feature parity with Orkes core scheduling
2. Orkes creates thin wrapper that injects orgId from request context
3. Orkes deprecates custom scheduler implementation
4. Shared core reduces divergence

## Future Enhancements

- Search API (filter by workflow, status, etc.)
- Tags system
- Bulk operations
- UI integration
- Schedule dependencies
- Metrics/monitoring

## References

- Orkes Conductor scheduler implementation: https://github.com/orkes-io/orkes-conductor
- Spring `CronExpression`: https://docs.spring.io/spring-framework/docs/current/javadoc-api/org/springframework/scheduling/support/CronExpression.html