schedule_node.ex

defmodule ScheduleNode do
	use GenServer

	# Replaces external cron. All scheduled jobs run from here.
	# Does NOT call impulse can_work. Scheduled jobs must run regardless of human activity.
	# Does NOT spawn AI models. Pure script execution.
	#
	# Job cadence is driven by wall-clock last-run timestamps (seeded to the
	# process start time), not a per-tick counter. That keeps every cadence
	# correct for any tick_minutes value and stops a restart from firing every
	# job at once. A restart resets the phase (each job waits a full interval
	# from startup), which is the accepted tradeoff for keeping this in memory.


	def start_link(opts) do
		GenServer.start_link(__MODULE__, opts, name: :schedule_node)
	end

	def init(_opts) do
		settings_path = Path.join(__DIR__, "settings.txt")
		content = read_file(settings_path)
		settings = case Jason.decode(content) do
			{:ok, decoded} -> decoded
			{:error, _} -> %{}
		end

		homepath = Map.get(settings, "homepath", "/app/")
		config = Map.get(settings, "schedule_pipeline", %{})
		tick_ms = compute_tick_ms(config, "tick_minutes", 60, 60000)

		File.mkdir_p(homepath <> "work/reports/")
		File.mkdir_p(homepath <> "work/reports/archive")
		File.mkdir_p(homepath <> "work/evidence/")

		{:ok, tref} = :timer.send_interval(tick_ms, :tick)

		now = :os.system_time(:second)

		{:ok, %{
			settings: settings,
			homepath: homepath,
			last_hourly: now,
			last_maintenance: now,
			last_ml: now,
			last_cron: now,
			last_consolidation: now,
			last_ollama: now,
			tasks: [],
			timer_ref: tref,
			tick_ms: tick_ms
		}}
	end


	# ===== THE MAIN LOOP =====

	def handle_info(:tick, state) do

		# kill any old tasks from the previous tick
		kill_tasks(state.tasks)
		state = %{state | tasks: []}

		state = reload_config(state)
		now = :os.system_time(:second)

		{state, tasks} = run_due_jobs(state, now)

		write_status(state, tasks)

		flush_ticks()
		{:noreply, %{state | tasks: tasks}}
	end


	# ===== ASYNC RESULT HANDLERS =====

	def handle_info({ref, _result}, state) when is_reference(ref) do
		Process.demonitor(ref, [:flush])
		{:noreply, state}
	end

	def handle_info({:DOWN, _ref, :process, _pid, _reason}, state) do
		{:noreply, state}
	end


	# ===== RECEIVING FROM OTHER NODES =====

	def handle_cast({:core_message, _from, _data}, state) do
		{:noreply, state}
	end

	def handle_cast({:master_message, _data}, state) do
		{:noreply, state}
	end


	# ===== JOB SCHEDULING =====
	# Each job runs when its wall-clock interval has elapsed since its last run.
	# Intervals in seconds: hourly 3600, 6h 21600, 12h 43200, 24h 86400,
	# weekly 604800. last_* is advanced only when the job actually runs.

	defp run_due_jobs(state, now) do
		hp = state.homepath

		{state, tasks} = maybe_run(state, :last_hourly, 3600, now, [], fn -> [spawn_hourly_coordinator(hp)] end)

		{state, tasks} = maybe_run(state, :last_maintenance, 21600, now, tasks, fn -> maintenance_jobs(hp) end)

		{state, tasks} = maybe_run(state, :last_ml, 43200, now, tasks, fn ->
			[Task.Supervisor.async_nolink(Ala.TaskSupervisor, fn ->
				System.cmd("python3", [hp <> "MLModels/run_training_cycle.py"], cd: hp <> "MLModels")
			end)]
		end)

		{state, tasks} = maybe_run(state, :last_cron, 86400, now, tasks, fn -> cron_jobs(hp) end)

		{state, tasks} = maybe_run(state, :last_consolidation, 21600, now, tasks, fn -> consolidation_jobs(hp) end)

		# weekly Ollama training is additionally gated by the trainOllama setting
		{state, tasks} = if Map.get(state.settings, "trainOllama", false) do
			maybe_run(state, :last_ollama, 604800, now, tasks, fn -> ollama_jobs(hp) end)
		else
			{state, tasks}
		end

		{state, tasks}
	end

	# Runs job_fun (which returns a list of tasks) when `interval` seconds have
	# elapsed since state[key], advancing state[key] to now only if it runs.
	defp maybe_run(state, key, interval, now, tasks, job_fun) do
		if now - Map.get(state, key) >= interval do
			{Map.put(state, key, now), tasks ++ job_fun.()}
		else
			{state, tasks}
		end
	end


	# ===== HOURLY COORDINATOR (off the GenServer) =====
	# Reports must finish before evidence runs, but that ordering must not block
	# the node loop. One coordinator task runs the report->evidence sequence and
	# awaits inside itself, so the GenServer stays responsive to casts/config.

	defp spawn_hourly_coordinator(hp) do
		Task.Supervisor.async_nolink(Ala.TaskSupervisor, fn ->
			# report pulls run BEFORE evidence so reports are fresh
			report_tasks = report_jobs(hp)
			await_all(report_tasks, 120_000)

			# evidence cross-reference for each type that has a matcher
			evidence_tasks = Enum.map(["marketing", "socialmedia", "customerservice", "devteam"], fn t ->
				Task.Supervisor.async_nolink(Ala.TaskSupervisor, fn ->
					System.cmd("python3", [hp <> "tools/evidence_cross_reference.py", hp, t], [])
				end)
			end)
			await_all(evidence_tasks, 120_000)
			:ok
		end)
	end

	defp await_all(tasks, timeout) do
		Enum.each(tasks, fn t ->
			try do
				Task.await(t, timeout)
			catch
				:exit, _ -> :ok
			end
		end)
	end


	# ===== TYPE-SPECIFIC REPORT JOBS =====
	# Each entry replaces a section of cron/hourly.sh

	defp report_jobs(hp) do
		reports = hp <> "work/reports/"
		[
			Task.Supervisor.async_nolink(Ala.TaskSupervisor, fn ->
				run_report("php", [hp <> "tools/ga4_report.php", "7"], reports <> "ga4_report", "campaigns")
			end),
			Task.Supervisor.async_nolink(Ala.TaskSupervisor, fn ->
				run_report("php", [hp <> "tools/gsc_report.php", "28", "keywords"], reports <> "gsc_report", "rows")
			end),
			Task.Supervisor.async_nolink(Ala.TaskSupervisor, fn ->
				System.cmd("python3", [hp <> "tools/socialmedia.py", "analytics", "all"], [])
			end)
		]
	end


	# ===== MAINTENANCE JOBS =====

	defp maintenance_jobs(hp) do
		[
			Task.Supervisor.async_nolink(Ala.TaskSupervisor, fn ->
				System.cmd("php", [hp <> "tools/customerservice.php", "maintenance", "full"], [])
			end),
			Task.Supervisor.async_nolink(Ala.TaskSupervisor, fn ->
				System.cmd("php", [hp <> "tools/marketing.php", "maintenance", "full"], [])
			end),
			Task.Supervisor.async_nolink(Ala.TaskSupervisor, fn ->
				System.cmd("python3", [hp <> "tools/socialmedia.py", "maintenance"], [])
			end),
			Task.Supervisor.async_nolink(Ala.TaskSupervisor, fn ->
				System.cmd("php", [hp <> "tools/devteam.php", "maintenance"], [])
			end)
		]
	end


	# ===== DAILY CRON SCRIPTS =====

	defp cron_jobs(hp) do
		case File.ls(hp <> "cron/") do
			{:ok, files} ->
				files
				|> Enum.filter(&String.ends_with?(&1, ".sh"))
				|> Enum.map(fn script ->
					Task.Supervisor.async_nolink(Ala.TaskSupervisor, fn ->
						System.cmd("bash", [hp <> "cron/" <> script], stderr_to_stdout: true)
					end)
				end)
			_ -> []
		end
	end


	# ===== TOPIC CONSOLIDATION =====

	defp consolidation_jobs(hp) do
		script = hp <> "tools/topic_consolidation.py"
		if File.exists?(script) do
			[Task.Supervisor.async_nolink(Ala.TaskSupervisor, fn ->
				System.cmd("python3", [script, hp], stderr_to_stdout: true)
			end)]
		else
			[]
		end
	end


	# ===== OLLAMA TRAINING (weekly) =====

	defp ollama_jobs(hp) do
		script = hp <> "tools/ollama_training.py"
		if File.exists?(script) do
			[Task.Supervisor.async_nolink(Ala.TaskSupervisor, fn ->
				System.cmd("python3", [script, "full", "--base-model", "mistral:7b"], cd: hp, stderr_to_stdout: true)
			end)]
		else
			[]
		end
	end


	# ===== REPORT HELPER =====
	# Same tmp/validate/mv pattern as cron/hourly.sh

	defp run_report(cmd, args, base_path, validate_key) do
		tmp_path = base_path <> ".tmp"
		out_path = base_path <> ".json"

		case System.cmd(cmd, args, []) do
			{output, 0} ->
				File.write(tmp_path, output)
				if String.contains?(output, "\"" <> validate_key <> "\"") do
					File.rename(tmp_path, out_path)
					# archive with date
					archive_dir = Path.dirname(out_path) <> "/archive"
					File.mkdir_p(archive_dir)
					date_str = Calendar.strftime(DateTime.utc_now(), "%Y%m%d")
					File.cp(out_path, archive_dir <> "/" <> Path.basename(base_path) <> "_" <> date_str <> ".json")
				else
					File.rm(tmp_path)
				end
			_ ->
				File.rm(tmp_path)
		end
	end


	# ===== STATUS =====

	defp write_status(state, tasks) do
		data = %{
			last_tick: :os.system_time(:second),
			jobs_this_tick: length(tasks),
			next_hourly: state.last_hourly + 3600
		}
		case Jason.encode(data) do
			{:ok, json} -> File.write(state.homepath <> "threads/schedulestatus.json", json)
			_ -> :ok
		end
	end


	# ===== CONFIG RELOAD =====

	defp reload_config(state) do
		content = read_file(Path.join(__DIR__, "settings.txt"))
		settings = case Jason.decode(content) do
			{:ok, decoded} -> decoded
			{:error, _} -> state.settings
		end

		config = Map.get(settings, "schedule_pipeline", %{})
		new_tick_ms = compute_tick_ms(config, "tick_minutes", 60, 60000)

		updated = if new_tick_ms != state.tick_ms do
			:timer.cancel(state.timer_ref)
			{:ok, new_tref} = :timer.send_interval(new_tick_ms, :tick)
			%{state | timer_ref: new_tref, tick_ms: new_tick_ms}
		else
			state
		end

		%{updated | settings: settings}
	end


	# ===== UTILITY FUNCTIONS =====

	defp kill_tasks(tasks) do
		Enum.each(tasks, fn t ->
			Process.demonitor(t.ref, [:flush])
			if Process.alive?(t.pid), do: Task.Supervisor.terminate_child(Ala.TaskSupervisor, t.pid)
		end)
	end

	# Clamp a configured tick to at least 1 unit, using the default for missing
# or non-numeric values, so a tick of 0, negative, or a bad type can never
# make send_interval flood the mailbox or crash the node on restart.
defp compute_tick_ms(config, key, default_units, ms_per_unit) do
	units = case Map.get(config, key) do
		n when is_number(n) -> max(n, 1)
		_ -> default_units
	end
	trunc(units * ms_per_unit)
end

defp read_file(path) do
		case File.read(path) do
			{:ok, content} -> content
			{:error, _} -> ""
		end
	end

	defp flush_ticks do
		receive do
			:tick -> flush_ticks()
		after
			0 -> :ok
		end
	end

end