Packages

condukt

0.16.5
1.7.0 1.6.3 1.6.2 1.6.1 1.6.0 1.5.1 1.5.0 1.4.1 1.4.0 1.3.0 1.2.1 1.2.0 1.1.0 1.0.0 0.21.0 0.20.0 0.19.0 0.18.0 0.17.0 0.16.6 0.16.5 0.16.4 0.16.3 0.16.2 0.16.1 0.16.0 0.15.1 0.15.0 0.14.2 0.14.1 0.14.0 0.13.1 0.13.0 0.12.0 0.11.0 0.10.0 0.9.0 0.8.0 0.7.2 0.7.1 0.7.0 0.6.1 0.6.0 0.5.0

A framework for building AI agents in Elixir

HexDocs

Current section

Files

Jump to
Readme
44 Versions
10 Dependencies
Dependants
Files
Activity
condukt lib condukt workflows runtime cron.ex
Raw

lib/condukt/workflows/runtime/cron.ex 

defmodule Condukt.Workflows.Runtime.Cron do

  @moduledoc """

  Cron trigger process for workflow runtimes.

  """



  use GenServer



  alias Condukt.Workflows.Runtime.Worker

  alias Crontab.CronExpression.Parser

  alias Crontab.Scheduler



  @doc false

  def start_link(opts), do: GenServer.start_link(__MODULE__, opts)



  @impl true

  def init(opts) do

    expr = Keyword.fetch!(opts, :expr)

    cron = Parser.parse!(expr, extended?(expr))



    state = %{

      workflow: Keyword.fetch!(opts, :workflow),

      cron: cron,

      input: Keyword.get(opts, :input, %{}),

      clock: Keyword.get(opts, :clock, &NaiveDateTime.utc_now/0),

      run_fun: Keyword.get(opts, :run_fun, &Worker.invoke/2),

      timer_ref: nil

    }



    {:ok, schedule_next(state)}

  end



  @impl true

  def handle_info(:fire, state) do

    %{workflow: workflow, input: input, run_fun: run_fun} = state

    Task.start(fn -> run_fun.(workflow.name, input) end)

    {:noreply, schedule_next(%{state | timer_ref: nil})}

  end



  defp schedule_next(%{cron: cron, clock: clock} = state) do

    now = clock.()

    next_run = Scheduler.get_next_run_date!(cron, now)

    delay = max(0, NaiveDateTime.diff(next_run, now, :millisecond))

    %{state | timer_ref: Process.send_after(self(), :fire, delay)}

  end



  defp extended?(expr) do

    expr

    |> String.split(" ", trim: true)

    |> length()

    |> Kernel.==(6)

  end

end