feat: cron

.rubocop.yml

@@ -47,6 +47,9 @@ Metrics/AbcSize:
   Exclude:
     - spec/**/*
 
+Metrics/ParameterLists:
+  Enabled: false
+
 Style/OneClassPerFile:
   Exclude:
     - spec/**/*

lib/cosmo/api.rb

@@ -4,6 +4,7 @@
 require "cosmo/api/counter"
 require "cosmo/api/kv"
 require "cosmo/api/stats"
+require "cosmo/api/cron"
 
 module Cosmo
   module API

lib/cosmo/api/cron.rb

@@ -0,0 +1,118 @@
+# frozen_string_literal: true
+
+require "securerandom"
+require "cosmo/api/cron/entry"
+
+module Cosmo
+  module API
+    # Web-facing API for cron schedules. Single interface for all cron NATS operations.
+    #
+    # Derives the schedule list entirely from NATS.
+    # Whatever is deployed in NATS is exactly what appears in the UI.
+    #
+    # Schedule templates live in the same job stream they target (e.g. +default+),
+    # stored at subjects matching +cosmo.cron.<stream>.>+. NATS 2.14 fires each
+    # template by publishing the body to +Nats-Schedule-Target+ as a regular
+    # JetStream message that accumulates alongside pending jobs.
+    class Cron
+      def self.instance
+        @instance ||= new
+      end
+
+      # @return [Array<Hash>] every cron schedule currently deployed in NATS
+      def all
+        Stream.jobs.flat_map { |s| schedules_from_stream(s.name) }
+      rescue StandardError
+        []
+      end
+
+      # Publish (or replace) a schedule message in NATS.
+      # @return [Hash, nil] the persisted schedule as a hash, or nil on failure
+      def upsert!(class_name: nil, stream: nil, schedule: nil, args: [], timezone: nil, name: nil)
+        e = Entry.new(class_name: class_name, stream: stream, expression: schedule,
+                      args: args, timezone: timezone, name: name)
+        headers = {
+          "Nats-Schedule" => e.expression,
+          "Nats-Schedule-Target" => e.target_subject
+        }
+        headers["Nats-Schedule-Time-Zone"] = e.timezone if e.timezone
+        client.publish(e.schedule_subject, e.job_payload, stream: e.stream, header: headers)
+        build_from_nats(e.stream, e.schedule_subject)
+      end
+
+      # Purge the schedule message from NATS (stops future firings).
+      # @param subject [String]
+      def delete!(subject)
+        stream_name = subject.to_s.split(".")[2]
+        client.purge(stream_name, subject)
+      rescue NATS::JetStream::Error::NotFound, NATS::IO::Timeout
+        nil
+      end
+
+      # Dispatch the job immediately to the target stream, bypassing the timer.
+      # @param schedule_subject [String] e.g. "cosmo.cron.default.report_job.daily"
+      def run_now!(schedule_subject) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
+        stream_name = schedule_subject.to_s.split(".")[2]
+        msg = client.get_message(stream_name, subject: schedule_subject)
+        return unless msg
+
+        headers = msg.headers || {}
+        body = Utils::Json.parse(msg.data) || {}
+        target = headers["Nats-Schedule-Target"]
+        return unless target && body[:class]
+
+        payload = Utils::Json.dump({
+                                     jid: SecureRandom.hex(12),
+                                     class: body[:class],
+                                     args: body[:args] || [],
+                                     retry: body[:retry] || Job::Data::DEFAULTS[:retry],
+                                     dead: body[:dead].nil? ? Job::Data::DEFAULTS[:dead] : body[:dead]
+                                   })
+        client.publish(target, payload, stream: stream_name)
+      rescue NATS::JetStream::Error::NotFound
+        nil
+      end
+
+      private
+
+      def client
+        @client ||= Client.instance
+      end
+
+      def schedules_from_stream(stream_name)
+        filter = "#{Entry::SUBJECT_PREFIX}.#{stream_name}.>"
+        subjects = client.cron_subjects_in_stream(stream_name, filter)
+        subjects.filter_map { |subj| build_from_nats(stream_name, subj) }
+      end
+
+      def build_from_nats(stream_name, subject)
+        msg = client.get_message(stream_name, subject: subject)
+        return unless msg
+
+        headers = msg.headers || {}
+        body = Utils::Json.parse(msg.data) || {}
+
+        {
+          class: body[:class],
+          stream: stream_name,
+          schedule: headers["Nats-Schedule"],
+          timezone: headers["Nats-Schedule-Time-Zone"],
+          args: body[:args] || [],
+          name: name_from_subject(subject),
+          schedule_subject: subject,
+          target_subject: headers["Nats-Schedule-Target"],
+          registry_key: subject.split(".").drop(2).join("/")
+        }
+      rescue StandardError
+        nil
+      end
+
+      # "cosmo.cron.default.report_job"         → nil
+      # "cosmo.cron.default.report_job.monthly" → "monthly"
+      def name_from_subject(subject)
+        parts = subject.to_s.split(".")
+        parts.length > 4 ? parts.drop(4).join(".") : nil
+      end
+    end
+  end
+end

lib/cosmo/api/cron/entry.rb

@@ -0,0 +1,99 @@
+# frozen_string_literal: true
+
+require "securerandom"
+require "json"
+
+module Cosmo
+  module API
+    class Cron
+      # Value object representing a single cron schedule entry.
+      # Each schedule maps one job class (and optional args) to a NATS 2.14 message schedule.
+      #
+      # NATS 2.14 message scheduling uses a 6-field cron format:
+      #   second minute hour day-of-month month day-of-week
+      #
+      # @-shortcuts (@daily, @every 5m, @at ...) are passed through unchanged —
+      # the server understands them natively. Plain 5-field UNIX cron expressions
+      # need a seconds field prepended to become valid 6-field expressions.
+      #
+      #   "0 9 * * 1-5"  →  "0 0 9 * * 1-5"   (at 09:00 on weekdays)
+      #   "@daily"        →  "@daily"            (unchanged)
+      class Entry
+        SUBJECT_PREFIX = "cosmo.cron"
+
+        def self.normalize_expression(expr)
+          str = expr.to_s.strip
+          return str if str.start_with?("@")
+
+          fields = str.split
+          fields.size == 5 ? "0 #{str}" : str
+        end
+
+        attr_reader :class_name, :stream, :expression, :args, :timezone, :name
+
+        # @param class_name [String] Fully-qualified Ruby class name (e.g. "ReportJob")
+        # @param stream     [String] Target job stream name (e.g. "default")
+        # @param expression [String] NATS schedule expression (@daily, @every 5m, "0 0 9 * * 1-5", etc.)
+        # @param args       [Array]  Arguments passed to the job's +perform+ method
+        # @param timezone   [String, nil] IANA timezone name (e.g. "America/New_York"). Cron expressions only.
+        # @param name       [String, Symbol, nil] Disambiguates multiple schedules on the same class.
+        def initialize(class_name:, stream:, expression:, args: [], timezone: nil, name: nil)
+          @class_name = class_name.to_s
+          @stream     = stream.to_s
+          @expression = self.class.normalize_expression(expression)
+          @args       = Array(args)
+          @timezone   = timezone
+          @name       = name&.to_s
+        end
+
+        # Subject where the schedule message lives in NATS (one per unique schedule).
+        # e.g. "cosmo.cron.default.report_job" or "cosmo.cron.default.report_job.monthly"
+        def schedule_subject
+          parts = [SUBJECT_PREFIX, @stream, job_name]
+          parts << @name if @name
+          parts.join(".")
+        end
+
+        # Subject where NATS fires the generated job message.
+        # Must be a subject covered by the same stream.
+        def target_subject
+          "jobs.#{@stream}.#{job_name}"
+        end
+
+        def job_name
+          @job_name ||= Utils::String.underscore(@class_name)
+        end
+
+        def as_json
+          {
+            class: @class_name,
+            stream: @stream,
+            schedule: @expression,
+            timezone: @timezone,
+            args: @args,
+            name: @name,
+            schedule_subject: schedule_subject,
+            target_subject: target_subject
+          }.compact
+        end
+
+        # JSON payload sent as the body of the schedule message.
+        # Mirrors the format produced by Job::Data so the job processor can handle it.
+        def job_payload
+          Utils::Json.dump({
+                             jid: SecureRandom.hex(12),
+                             class: @class_name,
+                             args: @args,
+                             retry: ::Cosmo::Job::Data::DEFAULTS[:retry],
+                             dead: ::Cosmo::Job::Data::DEFAULTS[:dead]
+                           })
+        end
+
+        def to_s
+          "#<Cosmo::API::Cron::Entry class=#{@class_name} expression=#{@expression} stream=#{@stream}>"
+        end
+        alias inspect to_s
+      end
+    end
+  end
+end

lib/cosmo/api/stream.rb

@@ -35,7 +35,9 @@ def info
       end
 
       def total
-        info[:state].messages.to_i
+        all_msgs = info[:state].messages.to_i
+        cron_count = Client.instance.cron_subjects_in_stream(name, "#{Cron::Entry::SUBJECT_PREFIX}.#{name}.>").size
+        [all_msgs - cron_count, 0].max
       rescue NATS::Error
         0
       end
@@ -84,7 +86,10 @@ def messages(page: nil, limit: nil)
       end
 
       def message(seq)
-        Job.new(name, client.get_message(name, seq: seq, direct: true))
+        job = Job.new(name, client.get_message(name, seq: seq, direct: true))
+        return if job.subject.to_s.start_with?(Cron::Entry::SUBJECT_PREFIX)
+
+        job
       rescue NATS::JetStream::Error::NotFound
         # nop, acked/nacked
       end

lib/cosmo/cli.rb

@@ -4,6 +4,7 @@
 require "optparse"
 
 module Cosmo
+  # rubocop:disable Metrics/AbcSize, Metrics/MethodLength, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity, Metrics/BlockLength
   class CLI
     def self.run
       instance.run
@@ -75,8 +76,8 @@ def require_path(path)
       require_files("app/streams") if File.directory?("app/streams")
     end
 
-    def flags_parser(flags) # rubocop:disable Metrics/AbcSize, Metrics/MethodLength
-      OptionParser.new do |o| # rubocop:disable Metrics/BlockLength
+    def flags_parser(flags)
+      OptionParser.new do |o|
         o.banner = "Usage: cosmo [flags] [command] [options]"
         o.separator ""
         o.separator "Command:"
@@ -102,20 +103,31 @@ def flags_parser(flags) # rubocop:disable Metrics/AbcSize, Metrics/MethodLength
           flags[:config_file] = arg
         end
 
-        o.on "-S", "--setup", "Load config, create streams and exit" do
+        o.on "-S", "--setup", "Create/update streams and sync cron schedules, then exit" do
           load_config(flags)
           boot_application
 
-          Config[:setup]&.each_value do |configs|
+          Config[:setup]&.each do |type, configs|
+            next if type == :cron
+
+            first_line = true
             configs.each do |name, config|
-              Client.instance.stream_info(name)
-            rescue NATS::JetStream::Error::NotFound
-              meta = { metadata: { "_cosmo.type" => "jobs" } }
-              Client.instance.create_stream(name, config.merge(meta))
+              meta = { metadata: { "_cosmo.type" => "jobs" } } if type == :jobs
+              Client.instance.setup_stream(name.to_s, config.merge(Hash(meta)))
+              first_line ? print("Stream is ready: #{name}") : print(", #{name}")
+              first_line = false
             end
           end
 
-          puts "Cosmo streams were created successfully"
+          puts
+          schedules = Config.dig(:setup, :cron)&.reduce(0) do |sum, (name, entry)|
+            class_name = entry.delete(:class)
+            API::Cron.instance.upsert!(**entry, name: name, class_name: class_name)
+            sum + 1
+          end
+
+          puts "Cron sync complete: #{schedules} schedule(s) registered" unless schedules.zero?
+          puts "Cosmo streams#{" and cron schedules" unless schedules.zero?} set up successfully."
           exit(0)
         end
 
@@ -238,4 +250,5 @@ def self.banner
     end
     # rubocop:enable Layout/TrailingWhitespace,Lint/IneffectiveAccessModifier
   end
+  # rubocop:enable Metrics/AbcSize, Metrics/MethodLength, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity, Metrics/BlockLength
 end

lib/cosmo/client.rb

@@ -42,6 +42,27 @@ def update_stream(name, config)
       js.update_stream(name: name, **config)
     end
 
+    # Create/update a stream, falling back to create when there's no stream.
+    # @param name [String] Stream name
+    # @param config [Hash] Full desired stream configuration
+    def setup_stream(name, config)
+      update_stream(name, config)
+    rescue NATS::JetStream::Error::StreamNotFound
+      create_stream(name, config)
+    end
+
+    # Return all subjects in +stream_name+ that match +filter+ using NATS's
+    # subjects_filter on STREAM.INFO (requires NATS ≥ 2.9).
+    # @return [Array<String>]
+    def cron_subjects_in_stream(stream_name, filter)
+      payload = Utils::Json.dump({ subjects_filter: filter })
+      resp = nc.request("$JS.API.STREAM.INFO.#{stream_name}", payload)
+      data = Utils::Json.parse(resp.data, symbolize_names: false)
+      (data&.dig("state", "subjects") || {}).keys
+    rescue StandardError
+      []
+    end
+
     def list_streams
       response = nc.request("$JS.API.STREAM.LIST", "")
       data = Utils::Json.parse(response.data, symbolize_names: false)

lib/cosmo/config.rb

@@ -31,11 +31,20 @@ def self.normalize!(config) # rubocop:disable Metrics/AbcSize, Metrics/MethodLen
       end
 
       config[:setup]&.each_key do |type|
+        next if type == :cron
+
         config[:setup][type]&.each_key do |name|
           c = config[:setup][type][name]
           c[:max_age] = c[:max_age].to_i * NANO if c[:max_age]
           c[:duplicate_window] = c[:duplicate_window].to_i * NANO if c[:duplicate_window]
           c[:subjects] = c[:subjects].map { |s| format(s, name: name) } if c[:subjects]
+
+          next unless type == :jobs # Every jobs stream supports NATS 2.14 message scheduling.
+
+          c[:allow_msg_schedules] = true
+          cron_subject = "#{API::Cron::Entry::SUBJECT_PREFIX}.#{name}.>"
+          c[:subjects] = Array(c[:subjects])
+          c[:subjects] << cron_subject unless c[:subjects].include?(cron_subject)
         end
       end
     end

lib/cosmo/stream.rb

@@ -12,7 +12,7 @@ def self.included(base)
     end
 
     module ClassMethods
-      def options(stream: nil, consumer_name: nil, batch_size: nil, fetch_timeout: nil, start_position: nil, consumer: nil, publisher: nil) # rubocop:disable Metrics/ParameterLists
+      def options(stream: nil, consumer_name: nil, batch_size: nil, fetch_timeout: nil, start_position: nil, consumer: nil, publisher: nil)
         register
         default_options.merge!({ stream:, consumer_name:, batch_size:, fetch_timeout:, start_position:, consumer:, publisher: }.compact)
       end

lib/cosmo/utils/overrides.rb

@@ -1,7 +1,7 @@
 # frozen_string_literal: true
 
 Cosmo::Utils::Warnings.silence do
-  members = NATS::JetStream::API::StreamConfig.members + [:allow_msg_counter]
+  members = NATS::JetStream::API::StreamConfig.members + %i[allow_msg_counter allow_msg_schedules]
   NATS::JetStream::API::StreamConfig = Struct.new(*members, keyword_init: true) do
     def initialize(opts = {})
       rem = opts.keys - members