From b85e1204d2d6ec1ca5cc0ba7da4f92cdcff86d33 Mon Sep 17 00:00:00 2001
From: yujonglee <yujonglee.dev@gmail.com>
Date: Wed, 25 Sep 2024 13:27:02 +0900
Subject: [PATCH] move cancel_fetch to source resource

---
 core/lib/canary/sources/source.ex             | 44 ++++++++++++++++++-
 .../lib/canary_web/live/source_live/detail.ex | 25 +++++------
 2 files changed, 54 insertions(+), 15 deletions(-)

diff --git a/core/lib/canary/sources/source.ex b/core/lib/canary/sources/source.ex
index 174cd075..bbd9dfc0 100644
--- a/core/lib/canary/sources/source.ex
+++ b/core/lib/canary/sources/source.ex
@@ -3,8 +3,10 @@ defmodule Canary.Sources.Source do
     domain: Canary.Sources,
     data_layer: AshPostgres.DataLayer
 
-  alias Canary.Sources.Document
   require Ash.Query
+  require Ecto.Query
+
+  alias Canary.Sources.Document
 
   attributes do
     uuid_primary_key :id
@@ -103,9 +105,49 @@ defmodule Canary.Sources.Source do
       change {Ash.Resource.Change.CascadeDestroy, relationship: :events, action: :destroy}
     end
 
+    update :cancel_fetch do
+      require_atomic? false
+
+      change set_attribute(:state, :idle)
+
+      change fn changeset, _ctx ->
+        %{id: source_id, config: %Ash.Union{type: type}} = changeset.data
+
+        worker =
+          case type do
+            :webpage -> Canary.Workers.WebpageProcessor
+            :github_issue -> Canary.Workers.GithubIssueProcessor
+            :github_discussion -> Canary.Workers.GithubDiscussionProcessor
+          end
+          |> to_string()
+          |> String.trim_leading("Elixir.")
+
+        changeset
+        |> Ash.Changeset.after_action(fn _, record ->
+          result =
+            Oban.Job
+            |> Ecto.Query.where(worker: ^worker)
+            |> Ecto.Query.where([j], json_extract_path(j.args, ["source_id"]) == ^source_id)
+            |> Oban.cancel_all_jobs()
+
+          case result do
+            {:ok, _} ->
+              {:ok, record}
+
+            {:error, error} ->
+              IO.inspect(error)
+              {:ok, record}
+          end
+        end)
+      end
+    end
+
     update :fetch do
       require_atomic? false
 
+      change set_attribute(:state, :running)
+      change set_attribute(:last_fetched_at, &DateTime.utc_now/0)
+
       change fn changeset, _ctx ->
         %{id: source_id, config: config} = changeset.data
 
diff --git a/core/lib/canary_web/live/source_live/detail.ex b/core/lib/canary_web/live/source_live/detail.ex
index a61f0038..c6626710 100644
--- a/core/lib/canary_web/live/source_live/detail.ex
+++ b/core/lib/canary_web/live/source_live/detail.ex
@@ -2,8 +2,6 @@ defmodule CanaryWeb.SourceLive.Detail do
   use CanaryWeb, :live_component
   alias PrimerLive.Component, as: Primer
 
-  require Ecto.Query
-
   @impl true
   def render(assigns) do
     ~H"""
@@ -351,20 +349,19 @@ defmodule CanaryWeb.SourceLive.Detail do
 
   @impl true
   def handle_event("cancel", _, socket) do
-    worker =
-      case socket.assigns.source.config.type do
-        :webpage -> Canary.Workers.WebpageProcessor
-        :github_issue -> Canary.Workers.GithubIssueProcessor
-        :github_discussion -> Canary.Workers.GithubDiscussionProcessor
-      end
-      |> to_string()
-      |> String.trim_leading("Elixir.")
+    result =
+      socket.assigns.source
+      |> Ash.Changeset.for_update(:cancel_fetch, %{})
+      |> Ash.update()
 
-    Oban.Job
-    |> Ecto.Query.where(worker: ^worker)
-    |> Oban.cancel_all_jobs()
+    case result do
+      {:ok, _} ->
+        {:noreply, socket}
 
-    {:noreply, socket}
+      {:error, error} ->
+        IO.inspect(error)
+        {:noreply, socket}
+    end
   end
 
   defp transform_params(params) do