feat(scheduler): add schedules read-model projection (#383)

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>

Author: yordis

proto/trogonai/scheduler/schedules/projections/v1/delivery.proto

@@ -0,0 +1,33 @@
+edition = "2024";
+
+package trogonai.scheduler.schedules.projections.v1;
+
+import "google/protobuf/duration.proto";
+
+// Delivery is the read model's own copy of where a schedule publishes when it
+// fires.
+//
+// It mirrors the command-side delivery today but is defined here so the read
+// model owns its full storage shape and can evolve independently of the event
+// and command schemas.
+message Delivery {
+  oneof kind {
+    NatsMessage nats_message = 1;
+  }
+
+  message NatsMessage {
+    string subject = 1 [features.field_presence = LEGACY_REQUIRED];
+    google.protobuf.Duration ttl = 2;
+    Source source = 3;
+
+    message Source {
+      oneof kind {
+        LatestFromSubject latest_from_subject = 1;
+      }
+    }
+
+    message LatestFromSubject {
+      string subject = 1 [features.field_presence = LEGACY_REQUIRED];
+    }
+  }
+}

proto/trogonai/scheduler/schedules/projections/v1/message.proto

@@ -0,0 +1,21 @@
+edition = "2024";
+
+package trogonai.scheduler.schedules.projections.v1;
+
+import "trogon/content/v1alpha1/content.proto";
+
+// Message is the read model's own copy of the static payload and headers
+// recorded when a schedule was created.
+//
+// It mirrors the command-side message today but is defined here so the read
+// model owns its full storage shape and can evolve independently of the event
+// and command schemas.
+message Message {
+  trogon.content.v1alpha1.Content content = 1 [features.field_presence = LEGACY_REQUIRED];
+  repeated Header headers = 2;
+}
+
+message Header {
+  string name = 1 [features.field_presence = LEGACY_REQUIRED];
+  string value = 2 [features.field_presence = LEGACY_REQUIRED];
+}

proto/trogonai/scheduler/schedules/projections/v1/schedule.proto

@@ -0,0 +1,43 @@
+edition = "2024";
+
+package trogonai.scheduler.schedules.projections.v1;
+
+import "google/protobuf/duration.proto";
+import "google/protobuf/timestamp.proto";
+import "google/type/datetime.proto";
+
+// Schedule is the read model's own copy of the scheduling strategy recorded when
+// a schedule was created.
+//
+// It mirrors the command-side schedule definition today but is defined here so
+// the read model owns its full storage shape and can evolve independently of the
+// event and command schemas.
+message Schedule {
+  oneof kind {
+    At at = 1;
+    Every every = 2;
+    Cron cron = 3;
+    RRule rrule = 4;
+  }
+
+  message At {
+    google.protobuf.Timestamp at = 1 [features.field_presence = LEGACY_REQUIRED];
+  }
+
+  message Every {
+    google.protobuf.Duration every = 1 [features.field_presence = LEGACY_REQUIRED];
+  }
+
+  message Cron {
+    string expr = 1 [features.field_presence = LEGACY_REQUIRED];
+    google.type.TimeZone timezone = 2 [features.field_presence = LEGACY_REQUIRED];
+  }
+
+  message RRule {
+    google.protobuf.Timestamp dtstart = 1 [features.field_presence = LEGACY_REQUIRED];
+    string rrule = 2 [features.field_presence = LEGACY_REQUIRED];
+    google.type.TimeZone timezone = 3 [features.field_presence = LEGACY_REQUIRED];
+    repeated google.protobuf.Timestamp rdate = 4;
+    repeated google.protobuf.Timestamp exdate = 5;
+  }
+}

proto/trogonai/scheduler/schedules/projections/v1/schedule_projection.proto

@@ -0,0 +1,37 @@
+edition = "2024";
+
+package trogonai.scheduler.schedules.projections.v1;
+
+import "google/protobuf/timestamp.proto";
+import "trogonai/scheduler/schedules/projections/v1/delivery.proto";
+import "trogonai/scheduler/schedules/projections/v1/message.proto";
+import "trogonai/scheduler/schedules/projections/v1/schedule.proto";
+import "trogonai/scheduler/schedules/projections/v1/schedule_status.proto";
+
+// ScheduleProjection is the read-model projection of a schedule's current state,
+// folded from the schedule event stream and stored as the value of each entry in
+// the schedules KV bucket.
+//
+// It is a derived, rebuildable read view, so it defines its own status, schedule,
+// delivery, and message types rather than embedding the event and command protos.
+// That keeps this storage shape free to evolve under protobuf's field rules in
+// isolation from the schemas it is folded from.
+message ScheduleProjection {
+  // Stable schedule id; also the schedule stream id.
+  string schedule_id = 1 [features.field_presence = LEGACY_REQUIRED];
+  // Current lifecycle status (scheduled or paused).
+  ScheduleStatus status = 2 [features.field_presence = LEGACY_REQUIRED];
+  // True once a recurring schedule has run to exhaustion: it stays visible but
+  // will never fire again.
+  bool completed = 3;
+  // The next planned occurrence, if one is armed and pending.
+  google.protobuf.Timestamp next_occurrence_at = 4;
+  // The most recently recorded occurrence, if any has fired.
+  google.protobuf.Timestamp last_occurrence_at = 5;
+  // The schedule definition recorded at creation.
+  Schedule schedule = 6 [features.field_presence = LEGACY_REQUIRED];
+  // The delivery definition recorded at creation.
+  Delivery delivery = 7 [features.field_presence = LEGACY_REQUIRED];
+  // The static message payload and headers recorded at creation.
+  Message message = 8 [features.field_presence = LEGACY_REQUIRED];
+}

proto/trogonai/scheduler/schedules/projections/v1/schedule_status.proto

@@ -0,0 +1,23 @@
+edition = "2024";
+
+package trogonai.scheduler.schedules.projections.v1;
+
+// ScheduleStatus is the read model's own copy of a schedule's lifecycle status.
+//
+// It mirrors the command-side status today but is defined here so the read model
+// owns its full storage shape and can evolve under protobuf's field rules without
+// depending on the event or command schemas.
+message ScheduleStatus {
+  // A oneof gives each lifecycle state a place to grow without adding fields
+  // that are invalid for other states.
+  oneof kind {
+    Scheduled scheduled = 1;
+    Paused paused = 2;
+  }
+
+  // Scheduled means the schedule participates in future fire decisions.
+  message Scheduled {}
+
+  // Paused means the schedule keeps its definition without future fires.
+  message Paused {}
+}

rsworkspace/crates/trogon-scheduler/src/projections/mod.rs

@@ -0,0 +1,51 @@
+//! The schedules read-model projection (a NATS KV view of current schedules,
+//! folded from the schedule event stream).
+//!
+//! # Boundary
+//!
+//! This module is the **write side** only: it folds `v1` event protos straight
+//! into the stored `projections.v1.ScheduleProjection` proto and owns that storage layout
+//! (bucket, key scheme, checkpoint — see [`schedules::storage`]). It depends on
+//! the event/view protos, the shared error type, and the shared events stream —
+//! never on the read model or the queries. Reading the stored proto back out and
+//! decoding it into the read-model value objects callers see is the **read side**
+//! and lives entirely in [`crate::queries`].
+//!
+//! # Single active writer
+//!
+//! This projection must be driven by a single active instance, the same
+//! invariant the execution worker requires (see `processor::execution::worker`).
+//! The event log itself is safe under concurrent writers — per-subject
+//! optimistic concurrency serializes appends — but the KV writes here are
+//! read-modify-write without compare-and-swap, and [`catch_up_schedules_read_model`]
+//! rebuilds the bucket from the event log. With two instances writing the same
+//! bucket, a write can land on stale state, or a boot-time rebuild can overwrite
+//! a peer's newer write.
+//!
+//! Enforce this by deploying the scheduler as a single active instance (a single
+//! replica, a `Recreate` rollout, or leader election) — not with a
+//! projection-specific lock, since the worker already gates the whole service.
+//!
+//! ## Safety net for the rolling-restart overlap
+//!
+//! A rolling deploy briefly runs two instances. Most of the projection degrades
+//! gracefully, not corrupts, during that window:
+//! - Catch-up early-returns when the checkpoint is current, so a booting instance
+//!   does not rebuild (or reconcile) in steady state.
+//! - The checkpoint only advances contiguously, so a racing writer can stall it
+//!   (forcing a harmless re-fold) but never advance it past an unprojected event.
+//! - A live projection failure never fails the durable append and never advances
+//!   the checkpoint, so any divergence self-heals on the next clean start.
+//!
+//! The one step that genuinely depends on the single-writer invariant is the
+//! catch-up reconcile: it deletes any current row absent from the freshly folded
+//! state, so during an overlap it could delete a row a peer just created. That
+//! row is re-created by the peer's next event or restart.
+//!
+//! The residual is a transient stale (or briefly missing) schedule that resolves
+//! on the schedule's next event or the next restart.
+
+mod schedules;
+
+pub(crate) use schedules::storage;
+pub(crate) use schedules::{catch_up_schedules_read_model, project_appended_events};