From ce17386803221bf5b982c13895662d2b172eb4f7 Mon Sep 17 00:00:00 2001 From: Dmitry Vorotilin Date: Mon, 15 Jun 2026 17:33:14 +0300 Subject: [PATCH] feat: cron --- .rubocop.yml | 3 + lib/cosmo/api.rb | 1 + lib/cosmo/api/cron.rb | 118 +++++++++++++++++++++++++ lib/cosmo/api/cron/entry.rb | 99 +++++++++++++++++++++ lib/cosmo/api/stream.rb | 9 +- lib/cosmo/cli.rb | 31 +++++-- lib/cosmo/client.rb | 21 +++++ lib/cosmo/config.rb | 9 ++ lib/cosmo/stream.rb | 2 +- lib/cosmo/utils/overrides.rb | 2 +- lib/cosmo/web.rb | 5 ++ lib/cosmo/web/assets/app.css | 28 ++++++ lib/cosmo/web/controllers/crons.rb | 41 +++++++++ lib/cosmo/web/helpers/application.rb | 4 + lib/cosmo/web/views/actions/index.erb | 2 +- lib/cosmo/web/views/crons/_table.erb | 58 ++++++++++++ lib/cosmo/web/views/crons/index.erb | 10 +++ lib/cosmo/web/views/jobs/_tabs.erb | 6 ++ lib/cosmo/web/views/jobs/busy.erb | 2 +- lib/cosmo/web/views/jobs/dead.erb | 2 +- lib/cosmo/web/views/jobs/enqueued.erb | 2 +- lib/cosmo/web/views/jobs/index.erb | 2 +- lib/cosmo/web/views/jobs/scheduled.erb | 2 +- lib/cosmo/web/views/layout.erb | 2 +- sig/cosmo/api/cron.rbs | 25 ++++++ sig/cosmo/api/cron/entry.rbs | 30 +++++++ sig/cosmo/client.rbs | 4 + spec/cosmo/api/cron_spec.rb | 89 +++++++++++++++++++ spec/cosmo/cron_spec.rb | 75 ++++++++++++++++ spec/support/cosmo.yml | 12 +++ 30 files changed, 676 insertions(+), 20 deletions(-) create mode 100644 lib/cosmo/api/cron.rb create mode 100644 lib/cosmo/api/cron/entry.rb create mode 100644 lib/cosmo/web/controllers/crons.rb create mode 100644 lib/cosmo/web/views/crons/_table.erb create mode 100644 lib/cosmo/web/views/crons/index.erb create mode 100644 lib/cosmo/web/views/jobs/_tabs.erb create mode 100644 sig/cosmo/api/cron.rbs create mode 100644 sig/cosmo/api/cron/entry.rbs create mode 100644 spec/cosmo/api/cron_spec.rb create mode 100644 spec/cosmo/cron_spec.rb diff --git a/.rubocop.yml b/.rubocop.yml index d7d8c48..d00126c 100644 --- a/.rubocop.yml +++ b/.rubocop.yml @@ -47,6 +47,9 @@ Metrics/AbcSize: Exclude: - spec/**/* +Metrics/ParameterLists: + Enabled: false + Style/OneClassPerFile: Exclude: - spec/**/* diff --git a/lib/cosmo/api.rb b/lib/cosmo/api.rb index add6113..c5e53a2 100644 --- a/lib/cosmo/api.rb +++ b/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 diff --git a/lib/cosmo/api/cron.rb b/lib/cosmo/api/cron.rb new file mode 100644 index 0000000..7f620a9 --- /dev/null +++ b/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..>+. 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] 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 diff --git a/lib/cosmo/api/cron/entry.rb b/lib/cosmo/api/cron/entry.rb new file mode 100644 index 0000000..5c4aced --- /dev/null +++ b/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 + "#" + end + alias inspect to_s + end + end + end +end diff --git a/lib/cosmo/api/stream.rb b/lib/cosmo/api/stream.rb index abe87b8..bbcff86 100644 --- a/lib/cosmo/api/stream.rb +++ b/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 diff --git a/lib/cosmo/cli.rb b/lib/cosmo/cli.rb index bc88967..68d2f2d 100644 --- a/lib/cosmo/cli.rb +++ b/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 diff --git a/lib/cosmo/client.rb b/lib/cosmo/client.rb index 59e894c..c60eb0c 100644 --- a/lib/cosmo/client.rb +++ b/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] + 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) diff --git a/lib/cosmo/config.rb b/lib/cosmo/config.rb index d9339ab..adad66d 100644 --- a/lib/cosmo/config.rb +++ b/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 diff --git a/lib/cosmo/stream.rb b/lib/cosmo/stream.rb index ec5104f..23b2315 100644 --- a/lib/cosmo/stream.rb +++ b/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 diff --git a/lib/cosmo/utils/overrides.rb b/lib/cosmo/utils/overrides.rb index eb5d68b..2e53fe2 100644 --- a/lib/cosmo/utils/overrides.rb +++ b/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 diff --git a/lib/cosmo/web.rb b/lib/cosmo/web.rb index dcee4d8..8e45d31 100644 --- a/lib/cosmo/web.rb +++ b/lib/cosmo/web.rb @@ -9,6 +9,7 @@ require "cosmo/web/controllers/jobs" require "cosmo/web/controllers/streams" require "cosmo/web/controllers/actions" +require "cosmo/web/controllers/crons" module Cosmo class Web @@ -43,6 +44,10 @@ def call(env) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, M in [:get, "/streams/_info"] then [Controllers::Streams, :_info] in [:patch, "/streams/pause"] then [Controllers::Streams, :pause] in [:patch, "/streams/unpause"] then [Controllers::Streams, :unpause] + in [:get, "/crons"] then [Controllers::Crons, :index] + in [:get, "/crons/_table"] then [Controllers::Crons, :_table] + in [:delete, "/crons/delete"] then [Controllers::Crons, :delete] + in [:post, "/crons/run"] then [Controllers::Crons, :run_now] in [:get, "/actions"] then [Controllers::Actions, :index] in [:get, "/assets/htmx.min.js.gz"] then serve("htmx.2.0.8.min.js.gz", "application/javascript", diff --git a/lib/cosmo/web/assets/app.css b/lib/cosmo/web/assets/app.css index f88ba89..8549311 100644 --- a/lib/cosmo/web/assets/app.css +++ b/lib/cosmo/web/assets/app.css @@ -292,6 +292,34 @@ section > header { z-index: auto; } +/* ── Section tabs ───────────────────────────────────────────────────────── */ +.tabs { + display: flex; + gap: 0; + border-bottom: 2px solid var(--color-border); + margin-bottom: var(--space-3x); +} + +.tabs a { + padding: var(--space) var(--space-3x); + text-decoration: none; + color: var(--color-text-light); + border-bottom: 2px solid transparent; + margin-bottom: -2px; + font-weight: 500; + transition: color 0.15s, border-color 0.15s; +} + +.tabs a:hover { + color: var(--color-text); + border-bottom-color: var(--color-border); +} + +.tabs a.active { + color: var(--color-primary); + border-bottom-color: var(--color-primary); +} + section .nav { display: flex; gap: var(--space); diff --git a/lib/cosmo/web/controllers/crons.rb b/lib/cosmo/web/controllers/crons.rb new file mode 100644 index 0000000..9032914 --- /dev/null +++ b/lib/cosmo/web/controllers/crons.rb @@ -0,0 +1,41 @@ +# frozen_string_literal: true + +require "cosmo/web/controllers/application" + +module Cosmo + class Web + module Controllers + class Crons < Application + def index + content_for :title, "Crons" + ok render("crons/index", layout: true) + end + + def _table + ok render("crons/_table", { schedules: cron.all }) + end + + # Dispatch the job immediately, bypassing the schedule timer. + # Expects params["subject"] = the schedule subject stored in NATS. + def run_now + subject = Rack::Utils.unescape(params["subject"].to_s) + cron.run_now!(subject) + ok + end + + # Purge the schedule from NATS so it stops firing. + def delete + subject = Rack::Utils.unescape(params["subject"].to_s) + cron.delete!(subject) + _table + end + + private + + def cron + @cron ||= API::Cron.instance + end + end + end + end +end diff --git a/lib/cosmo/web/helpers/application.rb b/lib/cosmo/web/helpers/application.rb index ac55f30..2d5c2fe 100644 --- a/lib/cosmo/web/helpers/application.rb +++ b/lib/cosmo/web/helpers/application.rb @@ -73,6 +73,10 @@ def current_page?(path) request_path == path end + def path_prefix?(*values) + values.any? { |v| @request.path_info.start_with?(v) } + end + def referrer?(path) referrer_uri = URI(@request.referrer) referrer_path = referrer_uri.path diff --git a/lib/cosmo/web/views/actions/index.erb b/lib/cosmo/web/views/actions/index.erb index d695fd7..3658bde 100644 --- a/lib/cosmo/web/views/actions/index.erb +++ b/lib/cosmo/web/views/actions/index.erb @@ -2,6 +2,6 @@
-
None
+
Coming soon, stay tuned
diff --git a/lib/cosmo/web/views/crons/_table.erb b/lib/cosmo/web/views/crons/_table.erb new file mode 100644 index 0000000..5f35d73 --- /dev/null +++ b/lib/cosmo/web/views/crons/_table.erb @@ -0,0 +1,58 @@ +<% if @schedules.empty? -%> +
+ No cron schedules deployed in NATS.
+ Add a cron: section to cosmo.yml and run + cosmo --setup to publish them. +
+<% else -%> +
+ + + + + + + + + + + + + <% @schedules.each do |s| -%> + "> + + + + + + + + <% end -%> + +
JobExpressionTimezoneArgsStreamActions
+
<%= h(s[:class].to_s) %>
+ <% if s[:name] -%> + <%= h(s[:name].to_s) %> + <% end -%> +
+ <%= h(s[:schedule_subject].to_s) %> +
+ 		<%= h(s[:schedule].to_s) %><%= h(s[:timezone] || "UTC") %> + <% if s[:args]&.any? -%> + <%= h(s[:args].inspect) %> + <% else -%> + — + <% end -%> + <%= h(s[:stream].to_s) %> + ► Run Now + 🗑 Delete +
+
+<% end -%> diff --git a/lib/cosmo/web/views/crons/index.erb b/lib/cosmo/web/views/crons/index.erb new file mode 100644 index 0000000..5f1e020 --- /dev/null +++ b/lib/cosmo/web/views/crons/index.erb @@ -0,0 +1,10 @@ +
+
<%= render('jobs/_tabs') %>
+ +
+
Loading cron schedules…
+
+
diff --git a/lib/cosmo/web/views/jobs/_tabs.erb b/lib/cosmo/web/views/jobs/_tabs.erb new file mode 100644 index 0000000..20b7bfe --- /dev/null +++ b/lib/cosmo/web/views/jobs/_tabs.erb @@ -0,0 +1,6 @@ + diff --git a/lib/cosmo/web/views/jobs/busy.erb b/lib/cosmo/web/views/jobs/busy.erb index eeb05f2..ef82974 100644 --- a/lib/cosmo/web/views/jobs/busy.erb +++ b/lib/cosmo/web/views/jobs/busy.erb @@ -1,5 +1,5 @@
-
+
<%= render('jobs/_tabs') %>
-
+
<%= render('jobs/_tabs') %>
-
+
<%= render('jobs/_tabs') %>
-
+
<%= render('jobs/_tabs') %>
-
+
<%= render('jobs/_tabs') %>
🚀 Cosmo
  • - ">Jobs + ">Jobs
  • ">Streams diff --git a/sig/cosmo/api/cron.rbs b/sig/cosmo/api/cron.rbs new file mode 100644 index 0000000..465f697 --- /dev/null +++ b/sig/cosmo/api/cron.rbs @@ -0,0 +1,25 @@ +module Cosmo + module API + class Cron + def all: () -> ::Array[Hash[Symbol, untyped]] + + def upsert!: (?Cosmo::API::Cron::Entry schedule_obj, + ?class_name: ::String?, + ?stream: ::String?, + ?expression: ::String?, + ?args: ::Array[untyped], + ?timezone: ::String?, + ?name: ::String?) -> Hash[Symbol, untyped]? + + def delete!: (::String subject) -> void + + def run_now!: (::String schedule_subject) -> void + + private + + def schedules_from_stream: (::String stream_name) -> ::Array[Hash[Symbol, untyped]] + def build_from_nats: (::String stream_name, ::String subject) -> Hash[Symbol, untyped]? + def name_from_subject: (::String subject) -> ::String? + end + end +end diff --git a/sig/cosmo/api/cron/entry.rbs b/sig/cosmo/api/cron/entry.rbs new file mode 100644 index 0000000..678e886 --- /dev/null +++ b/sig/cosmo/api/cron/entry.rbs @@ -0,0 +1,30 @@ +module Cosmo + module API + class Cron + class Entry + SUBJECT_PREFIX: ::String + + attr_reader class_name: ::String + attr_reader stream: ::String + attr_reader expression: ::String + attr_reader args: ::Array[untyped] + attr_reader timezone: ::String? + attr_reader name: ::String? + + def self.normalize_expression: (untyped expr) -> ::String + + def initialize: (class_name: ::String, stream: ::String, expression: ::String, + ?args: ::Array[untyped], ?timezone: ::String?, ?name: (::String | ::Symbol)?) -> void + + def schedule_subject: () -> ::String + def target_subject: () -> ::String + def job_name: () -> ::String + def job_payload: () -> ::String + def as_json: () -> Hash[Symbol, untyped] + def to_s: () -> ::String + alias inspect to_s + end + end + end +end + diff --git a/sig/cosmo/client.rbs b/sig/cosmo/client.rbs index 21c1c30..19af725 100644 --- a/sig/cosmo/client.rbs +++ b/sig/cosmo/client.rbs @@ -23,6 +23,10 @@ module Cosmo def update_stream: (::String | Symbol name, Hash[Symbol, untyped] config) -> untyped + def setup_stream: (::String name, Hash[Symbol, untyped] config) -> untyped + + def cron_subjects_in_stream: (::String stream_name, ::String filter) -> Array[::String] + def list_streams: () -> Array[Hash[::String, untyped]] def pause_stream: (::String name) -> void diff --git a/spec/cosmo/api/cron_spec.rb b/spec/cosmo/api/cron_spec.rb new file mode 100644 index 0000000..a422e60 --- /dev/null +++ b/spec/cosmo/api/cron_spec.rb @@ -0,0 +1,89 @@ +# frozen_string_literal: true + +RSpec.describe Cosmo::API::Cron do + subject(:api) { described_class.new } + + describe "#upsert!" do + let(:schedule) do + { class_name: "ReportJob", stream: "default", schedule: "@daily", args: ["weekly"], timezone: "UTC", name: "daily_report" } + end + + it "publishes to the correct schedule subject" do + expect(client).to receive(:publish).with( + "cosmo.cron.default.report_job.daily_report", + anything, + stream: "default", + header: hash_including("Nats-Schedule" => "@daily", "Nats-Schedule-Target" => "jobs.default.report_job") + ) + api.upsert!(**schedule) + end + + it "includes timezone header when present" do + expect(client).to receive(:publish).with( + anything, anything, + stream: anything, + header: hash_including("Nats-Schedule-Time-Zone" => "UTC") + ) + api.upsert!(**schedule) + end + + it "omits timezone header when nil" do + expect(client).to receive(:publish).with( + anything, anything, + stream: anything, + header: satisfy { |h| !h.key?("Nats-Schedule-Time-Zone") } + ) + api.upsert!(class_name: "ReportJob", stream: "default", schedule: "@daily") + end + + it "accepts keyword arguments to build the schedule" do + expect(client).to receive(:publish).with( + "cosmo.cron.default.report_job", + anything, + stream: "default", + header: hash_including("Nats-Schedule" => "@hourly") + ) + api.upsert!(class_name: "ReportJob", stream: "default", schedule: "@hourly") + end + end + + describe "#delete!" do + it "purges the correct stream and subject" do + expect(client).to receive(:purge).with("default", "cosmo.cron.default.report_job") + api.delete!("cosmo.cron.default.report_job") + end + + it "returns nil when subject is not found" do + allow(client).to receive(:purge).and_raise(NATS::JetStream::Error::NotFound) + expect(api.delete!("cosmo.cron.default.report_job")).to be_nil + end + end + + describe "#run_now!" do + let(:subject_str) { "cosmo.cron.default.report_job.daily" } + let(:msg) do + double("msg", + headers: { "Nats-Schedule-Target" => "jobs.default.report_job" }, + data: Cosmo::Utils::Json.dump({ class: "ReportJob", args: [], retry: 3, dead: true })) + end + + before do + allow(client).to receive(:get_message).with("default", subject: subject_str).and_return(msg) + end + + it "publishes to the target subject" do + expect(client).to receive(:publish).with( + "jobs.default.report_job", + satisfy { |p| Cosmo::Utils::Json.parse(p)[:class] == "ReportJob" }, + stream: "default" + ) + api.run_now!(subject_str) + end + + it "does nothing when schedule message is not found" do + allow(client).to receive(:get_message).and_return(nil) + expect(client).not_to receive(:publish) + api.run_now!(subject_str) + end + end +end diff --git a/spec/cosmo/cron_spec.rb b/spec/cosmo/cron_spec.rb new file mode 100644 index 0000000..ac0c193 --- /dev/null +++ b/spec/cosmo/cron_spec.rb @@ -0,0 +1,75 @@ +# frozen_string_literal: true + +RSpec.describe Cosmo::API::Cron::Entry do + around do |example| + original_cron = Cosmo::Config.dig(:setup, :cron) + example.run + Cosmo::Config.set(:setup, :cron, original_cron) + end + + subject(:entry) do + described_class.new(class_name: "TestCronJob", stream: "default", expression: "@daily", + args: ["arg1"], timezone: "Europe/Amsterdam", name: "daily") + end + + it "builds the correct schedule subject" do + expect(entry.schedule_subject).to eq("cosmo.cron.default.test_cron_job.daily") + end + + it "builds the correct target subject" do + expect(entry.target_subject).to eq("jobs.default.test_cron_job") + end + + describe ".normalize_expression" do + it "leaves @-shortcuts unchanged" do + expect(described_class.normalize_expression("@daily")).to eq("@daily") + expect(described_class.normalize_expression("@every 30m")).to eq("@every 30m") + end + + it "prepends a seconds field to 5-field cron expressions" do + expect(described_class.normalize_expression("0 9 * * 1-5")).to eq("0 0 9 * * 1-5") + expect(described_class.normalize_expression("30 6 * * 0")).to eq("0 30 6 * * 0") + end + + it "leaves already-6-field expressions unchanged" do + expect(described_class.normalize_expression("0 0 9 * * 1-5")).to eq("0 0 9 * * 1-5") + end + end + + it "produces a valid JSON job payload" do + payload = Cosmo::Utils::Json.parse(entry.job_payload) + expect(payload[:class]).to eq("TestCronJob") + expect(payload[:args]).to eq(["arg1"]) + expect(payload[:retry]).to eq(Cosmo::Job::Data::DEFAULTS[:retry]) + expect(payload[:dead]).to eq(Cosmo::Job::Data::DEFAULTS[:dead]) + expect(payload[:jid]).to be_a(String).and have_attributes(length: 24) + end + + context "without a name" do + subject(:entry) { described_class.new(class_name: "TestCronJob", stream: "default", expression: "@hourly") } + + it "omits the name from the schedule subject" do + expect(entry.schedule_subject).to eq("cosmo.cron.default.test_cron_job") + end + end + + describe "#as_json" do + it "includes all relevant fields" do + json = entry.as_json + expect(json[:class]).to eq("TestCronJob") + expect(json[:stream]).to eq("default") + expect(json[:schedule]).to eq("@daily") + expect(json[:timezone]).to eq("Europe/Amsterdam") + expect(json[:args]).to eq(["arg1"]) + expect(json[:name]).to eq("daily") + expect(json[:schedule_subject]).to eq("cosmo.cron.default.test_cron_job.daily") + expect(json[:target_subject]).to eq("jobs.default.test_cron_job") + end + + it "omits nil fields" do + e = described_class.new(class_name: "TestCronJob", stream: "default", expression: "@hourly") + expect(e.as_json).not_to have_key(:timezone) + expect(e.as_json).not_to have_key(:name) + end + end +end diff --git a/spec/support/cosmo.yml b/spec/support/cosmo.yml index 5c3849d..0601d77 100644 --- a/spec/support/cosmo.yml +++ b/spec/support/cosmo.yml @@ -61,6 +61,18 @@ setup: max_msgs: 10000 max_age: 604800 # 7d description: Broken jobs (DLQ) + # Cron schedules — supported >= NATS 2.14 + # cron: + # daily_report: + # class: ReportJob + # schedule: "@daily" + # stream: default + # weekday_digest: + # class: DigestJob + # schedule: "0 9 * * 1-5" + # stream: default + # timezone: America/New_York + # args: [weekday] development: verbose: false